前言

在項目中，需要對每天千萬級的遊戲評論資訊進行詞頻統計，在生産者一端，我們将資料按照每天的拉取時間存入了Kafka當中，而在消費者一端，我們利用了spark streaming從kafka中不斷拉取資料進行詞頻統計。本文首先對spark streaming嵌入kafka的方式進行歸納總結，之後簡單闡述Spark streaming+kafka在輿情項目中的應用，最後将自己在Spark Streaming+kafka的實際優化中的一些經驗進行歸納總結。（如有任何纰漏歡迎補充來踩，我會第一時間改正^v^）

Spark streaming接收Kafka資料

用spark streaming流式處理kafka中的資料，第一步當然是先把資料接收過來，轉換為spark streaming中的資料結構Dstream。接收資料的方式有兩種：1.利用Receiver接收資料，2.直接從kafka讀取資料。

基于Receiver的方式

這種方式利用接收器（Receiver）來接收kafka中的資料，其最基本是使用Kafka高階使用者API接口。對于所有的接收器，從kafka接收來的資料會存儲在spark的executor中，之後spark streaming送出的job會處理這些資料。如下圖：

[圖檔上傳中...(image-d08014-1553529903049-7)]

在使用時，我們需要添加相應的依賴包：

<dependency><!-- Spark Streaming Kafka -->
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming-kafka_2.10</artifactId>
    <version>1.6.3</version>
</dependency>複制代碼

而對于Scala的基本使用方式如下：

import org.apache.spark.streaming.kafka._

 val kafkaStream = KafkaUtils.createStream(streamingContext, 
     [ZK quorum], [consumer group id], [per-topic number of Kafka partitions to consume])複制代碼

還有幾個需要注意的點：

在Receiver的方式中，Spark中的partition和kafka中的partition并不是相關的，是以如果我們加大每個topic的partition數量，僅僅是增加線程來處理由單一Receiver消費的主題。但是這并沒有增加Spark在處理資料上的并行度。
對于不同的Group和topic我們可以使用多個Receiver建立不同的Dstream來并行接收資料，之後可以利用union來統一成一個Dstream。
如果我們啟用了Write Ahead Logs複制到檔案系統如HDFS，那麼storage level需要設定成 StorageLevel.MEMORY_AND_DISK_SER，也就是 KafkaUtils.createStream(..., StorageLevel.MEMORY_AND_DISK_SER)

直接讀取方式

在spark1.3之後，引入了Direct方式。不同于Receiver的方式，Direct方式沒有receiver這一層，其會周期性的擷取Kafka中每個topic的每個partition中的最新offsets，之後根據設定的maxRatePerPartition來處理每個batch。其形式如下圖：

[圖檔上傳中...(image-6434a6-1553529903049-6)]

這種方法相較于Receiver方式的優勢在于：

簡化的并行：在Receiver的方式中我們提到建立多個Receiver之後利用union來合并成一個Dstream的方式提高資料傳輸并行度。而在Direct方式中，Kafka中的partition與RDD中的partition是一一對應的并行讀取Kafka資料，這種映射關系也更利于了解和優化。
高效：在Receiver的方式中，為了達到0資料丢失需要将資料存入Write Ahead Log中，這樣在Kafka和日志中就儲存了兩份資料，浪費！而第二種方式不存在這個問題，隻要我們Kafka的資料保留時間足夠長，我們都能夠從Kafka進行資料恢複。
精确一次：在Receiver的方式中，使用的是Kafka的高階API接口從Zookeeper中擷取offset值，這也是傳統的從Kafka中讀取資料的方式，但由于Spark Streaming消費的資料和Zookeeper中記錄的offset不同步，這種方式偶爾會造成資料重複消費。而第二種方式，直接使用了簡單的低階Kafka API，Offsets則利用Spark Streaming的checkpoints進行記錄，消除了這種不一緻性。

以上主要是對官方文檔[1]的一個簡單翻譯，詳細内容大家可以直接看下官方文檔這裡不再贅述。

不同于Receiver的方式，是從Zookeeper中讀取offset值，那麼自然zookeeper就儲存了目前消費的offset值，那麼如果重新啟動開始消費就會接着上一次offset值繼續消費。而在Direct的方式中，我們是直接從kafka來讀資料，那麼offset需要自己記錄，可以利用checkpoint、資料庫或檔案記錄或者回寫到zookeeper中進行記錄。這裡我們給出利用Kafka底層API接口，将offset及時同步到zookeeper中的通用類，我将其放在了github上：

Spark streaming+Kafka demo

示例中KafkaManager是一個通用類，而KafkaCluster是kafka源碼中的一個類，由于包名權限的原因我把它單獨提出來，ComsumerMain簡單展示了通用類的使用方法，在每次建立KafkaStream時，都會先從zooker中檢視上次的消費記錄offsets，而每個batch處理完成後，會同步offsets到zookeeper中。

Spark向kafka中寫入資料

上文闡述了Spark如何從Kafka中流式的讀取資料，下面我整理向Kafka中寫資料。與讀資料不同，Spark并沒有提供統一的接口用于寫入Kafka，是以我們需要使用底層Kafka接口進行包裝。

最直接的做法我們可以想到如下這種方式：

input.foreachRDD(rdd =>
  // 不能在這裡建立KafkaProducer
  rdd.foreachPartition(partition =>
    partition.foreach{
      case x:String=>{
        val props = new HashMap[String, Object]()
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
          "org.apache.kafka.common.serialization.StringSerializer")
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
          "org.apache.kafka.common.serialization.StringSerializer")
        println(x)
        val producer = new KafkaProducer[String,String](props)
        val message=new ProducerRecord[String, String]("output",null,x)
        producer.send(message)
      }
    }
  )
) 複制代碼

但是這種方式缺點很明顯，對于每個partition的每條記錄，我們都需要建立KafkaProducer，然後利用producer進行輸出操作，注意這裡我們并不能将KafkaProducer的建立任務放在foreachPartition外邊，因為KafkaProducer是不可序列化的（not serializable）。顯然這種做法是不靈活且低效的，因為每條記錄都需要建立一次連接配接。如何解決呢？

首先，我們需要将KafkaProducer利用lazy val的方式進行包裝如下：

import java.util.concurrent.Future
import org.apache.kafka.clients.producer.{ KafkaProducer, ProducerRecord, RecordMetadata }
class KafkaSink[K, V](createProducer: () => KafkaProducer[K, V]) extends Serializable {
  /* This is the key idea that allows us to work around running into
     NotSerializableExceptions. */
  lazy val producer = createProducer()
  def send(topic: String, key: K, value: V): Future[RecordMetadata] =
    producer.send(new ProducerRecord[K, V](topic, key, value))
  def send(topic: String, value: V): Future[RecordMetadata] =
    producer.send(new ProducerRecord[K, V](topic, value))
}

object KafkaSink {
  import scala.collection.JavaConversions._
  def apply[K, V](config: Map[String, Object]): KafkaSink[K, V] = {
    val createProducerFunc = () => {
      val producer = new KafkaProducer[K, V](config)
      sys.addShutdownHook {
        // Ensure that, on executor JVM shutdown, the Kafka producer sends
        // any buffered messages to Kafka before shutting down.
        producer.close()
      }
      producer
    }
    new KafkaSink(createProducerFunc)
  }
  def apply[K, V](config: java.util.Properties): KafkaSink[K, V] = apply(config.toMap)
}複制代碼

之後我們利用廣播變量的形式，将KafkaProducer廣播到每一個executor，如下：

// 廣播KafkaSink
val kafkaProducer: Broadcast[KafkaSink[String, String]] = {
  val kafkaProducerConfig = {
    val p = new Properties()
    p.setProperty("bootstrap.servers", Conf.brokers)
    p.setProperty("key.serializer", classOf[StringSerializer].getName)
    p.setProperty("value.serializer", classOf[StringSerializer].getName)
    p
  }
  log.warn("kafka producer init done!")
  ssc.sparkContext.broadcast(KafkaSink[String, String](kafkaProducerConfig))
}複制代碼

這樣我們就能在每個executor中愉快的将資料輸入到kafka當中：

//輸出到kafka
segmentedStream.foreachRDD(rdd => {
  if (!rdd.isEmpty) {
    rdd.foreach(record => {
      kafkaProducer.value.send(Conf.outTopics, record._1.toString, record._2)
      // do something else
    })
  }
})複制代碼

Spark streaming+Kafka應用

WeTest輿情監控對于每天爬取的千萬級遊戲玩家評論資訊都要實時的進行詞頻統計，對于爬取到的遊戲玩家評論資料，我們會生産到Kafka中，而另一端的消費者我們采用了Spark Streaming來進行流式處理，首先利用上文我們闡述的Direct方式從Kafka拉取batch，之後經過分詞、統計等相關處理，回寫到DB上（至于Spark中DB的回寫方式可參考我之前總結的博文：

Spark踩坑記——資料庫（Hbase+Mysql）

），由此高效實時的完成每天大量資料的詞頻統計任務。

Spark streaming+Kafka調優

Spark streaming+Kafka的使用中，當資料量較小，很多時候預設配置和使用便能夠滿足情況，但是當資料量大的時候，就需要進行一定的調整和優化，而這種調整和優化本身也是不同的場景需要不同的配置。

合理的批處理時間（batchDuration）

幾乎所有的Spark Streaming調優文檔都會提及批處理時間的調整，在StreamingContext初始化的時候，有一個參數便是批處理時間的設定。如果這個值設定的過短，即個batchDuration所産生的Job并不能在這期間完成處理，那麼就會造成資料不斷堆積，最終導緻Spark Streaming發生阻塞。而且，一般對于batchDuration的設定不會小于500ms，因為過小會導緻SparkStreaming頻繁的送出作業，對整個streaming造成額外的負擔。在平時的應用中，根據不同的應用場景和硬體配置，我設在1~10s之間，我們可以根據SparkStreaming的可視化監控界面，觀察Total Delay來進行batchDuration的調整，如下圖：

[圖檔上傳中...(image-b1fc91-1553529903048-5)]

合理的Kafka拉取量（maxRatePerPartition重要）

對于Spark Streaming消費kafka中資料的應用場景，這個配置是非常關鍵的，配置參數為：spark.streaming.kafka.maxRatePerPartition。這個參數預設是沒有上線的，即kafka當中有多少資料它就會直接全部拉出。而根據生産者寫入Kafka的速率以及消費者本身處理資料的速度，同時這個參數需要結合上面的batchDuration，使得每個partition拉取在每個batchDuration期間拉取的資料能夠順利的處理完畢，做到盡可能高的吞吐量，而這個參數的調整可以參考可視化監控界面中的Input Rate和Processing Time，如下圖：

[圖檔上傳中...(image-ca4e57-1553529903048-4)]

[圖檔上傳中...(image-e561ae-1553529903048-3)]

緩存反複使用的Dstream（RDD）

Spark中的RDD和SparkStreaming中的Dstream，如果被反複的使用，最好利用cache()，将該資料流緩存起來，防止過度的排程資源造成的網絡開銷。可以參考觀察Scheduling Delay參數，如下圖：

[圖檔上傳中...(image-f0679e-1553529903048-2)]

設定合理的GC

長期使用Java的小夥伴都知道，JVM中的垃圾回收機制，可以讓我們不過多的關注與記憶體的配置設定回收，更加專注于業務邏輯，JVM都會為我們搞定。對JVM有些了解的小夥伴應該知道，在Java虛拟機中，将記憶體分為了初生代（eden generation）、年輕代（young generation）、老年代（old generation）以及永久代（permanent generation），其中每次GC都是需要耗費一定時間的，尤其是老年代的GC回收，需要對記憶體碎片進行整理，通常采用标記-清楚的做法。同樣的在Spark程式中，JVM GC的頻率和時間也是影響整個Spark效率的關鍵因素。在通常的使用中建議：

--conf "spark.executor.extraJavaOptions=-XX:+UseConcMarkSweepGC"複制代碼

設定合理的CPU資源數

CPU的core數量，每個executor可以占用一個或多個core，可以通過觀察CPU的使用率變化來了解計算資源的使用情況，例如，很常見的一種浪費是一個executor占用了多個core，但是總的CPU使用率卻不高（因為一個executor并不總能充分利用多核的能力），這個時候可以考慮讓麼個executor占用更少的core，同時worker下面增加更多的executor，或者一台host上面增加更多的worker來增加并行執行的executor的數量，進而增加CPU使用率。但是增加executor的時候需要考慮好記憶體消耗，因為一台機器的記憶體配置設定給越多的executor，每個executor的記憶體就越小，以緻出現過多的資料spill over甚至out of memory的情況。

設定合理的parallelism

partition和parallelism，partition指的就是資料分片的數量，每一次task隻能處理一個partition的資料，這個值太小了會導緻每片資料量太大，導緻記憶體壓力，或者諸多executor的計算能力無法利用充分；但是如果太大了則會導緻分片太多，執行效率降低。在執行action類型操作的時候（比如各種reduce操作），partition的數量會選擇parent RDD中最大的那一個。而parallelism則指的是在RDD進行reduce類操作的時候，預設傳回資料的paritition數量（而在進行map類操作的時候，partition數量通常取自parent RDD中較大的一個，而且也不會涉及shuffle，是以這個parallelism的參數沒有影響）。是以說，這兩個概念密切相關，都是涉及到資料分片的，作用方式其實是統一的。通過spark.default.parallelism可以設定預設的分片數量，而很多RDD的操作都可以指定一個partition參數來顯式控制具體的分片數量。

在SparkStreaming+kafka的使用中，我們采用了Direct連接配接方式，前文闡述過Spark中的partition和Kafka中的Partition是一一對應的，我們一般預設設定為Kafka中Partition的數量。

使用高性能的算子

這裡參考了美團技術團隊的博文，并沒有做過具體的性能測試，其建議如下：

使用reduceByKey/aggregateByKey替代groupByKey
使用mapPartitions替代普通map
使用foreachPartitions替代foreach
使用filter之後進行coalesce操作
使用repartitionAndSortWithinPartitions替代repartition與sort類操作

使用Kryo優化序列化性能

這個優化原則我本身也沒有經過測試，但是好多優化文檔有提到，這裡也記錄下來。

在Spark中，主要有三個地方涉及到了序列化：

在算子函數中使用到外部變量時，該變量會被序列化後進行網絡傳輸（見“原則七：廣播大變量”中的講解）。
将自定義的類型作為RDD的泛型類型時（比如JavaRDD，Student是自定義類型），所有自定義類型對象，都會進行序列化。是以這種情況下，也要求自定義的類必須實作Serializable接口。
使用可序列化的持久化政策時（比如MEMORY_ONLY_SER），Spark會将RDD中的每個partition都序列化成一個大的位元組數組。

對于這三種出現序列化的地方，我們都可以通過使用Kryo序列化類庫，來優化序列化和反序列化的性能。Spark預設使用的是Java的序列化機制，也就是ObjectOutputStream/ObjectInputStream API來進行序列化和反序列化。但是Spark同時支援使用Kryo序列化庫，Kryo序列化類庫的性能比Java序列化類庫的性能要高很多。官方介紹，Kryo序列化機制比Java序列化機制，性能高10倍左右。Spark之是以預設沒有使用Kryo作為序列化類庫，是因為Kryo要求最好要注冊所有需要進行序列化的自定義類型，是以對于開發者來說，這種方式比較麻煩。

以下是使用Kryo的代碼示例，我們隻要設定序列化類，再注冊要序列化的自定義類型即可（比如算子函數中使用到的外部變量類型、作為RDD泛型類型的自定義類型等）：

// 建立SparkConf對象。
val conf = new SparkConf().setMaster(...).setAppName(...)
// 設定序列化器為KryoSerializer。
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
// 注冊要序列化的自定義類型。
conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))複制代碼

結果

經過種種調試優化，我們最終要達到的目的是，Spark Streaming能夠實時的拉取Kafka當中的資料，并且能夠保持穩定，如下圖所示：

[圖檔上傳中...(image-4afd1c-1553529903048-1)]

當然不同的應用場景會有不同的圖形，這是本文詞頻統計優化穩定後的監控圖，我們可以看到Processing Time這一柱形圖中有一Stable的虛線，而大多數Batch都能夠在這一虛線下處理完畢，說明整體Spark Streaming是運作穩定的。

Spark Streaming應該如何消費Kafka？前言Spark streaming接收Kafka資料Spark向kafka中寫入資料Spark streaming+Kafka應用Spark streaming+Kafka調優參考文獻

前言

Spark streaming接收Kafka資料

基于Receiver的方式

直接讀取方式

Spark向kafka中寫入資料

Spark streaming+Kafka應用

Spark streaming+Kafka調優

合理的批處理時間（batchDuration）

合理的Kafka拉取量（maxRatePerPartition重要）

緩存反複使用的Dstream（RDD）

設定合理的GC

設定合理的CPU資源數

設定合理的parallelism

使用高性能的算子

使用Kryo優化序列化性能

結果

參考文獻

繼續閱讀

用c++ 連接配接kafka我所踩過的坑(Connection refused || desired partition does not exist)

Linux 安裝kafka的庫librdkafka

librdkafka安裝步驟

Kafka：Topic概念與API介紹

Doris SQL日志審計部署，以及sql收集輸出kafka，後續血緣分析

Flink Kafka Doris實戰demo

延雲行業搜尋資料庫在大資料生态中位置和重要性大資料的挑戰大資料技術的現狀延雲行業搜尋資料庫

Spark在windows環境裡跑時報錯找不到org.apache.hadoop.fs.FSDataInputStream

Kafka 和 EMS 消息批量 ack 的實作

Spark流式分析系統實作流式實時日志分析系統

Spring Cloud整合Sleuth，當請求完成後，Zipkin沒有鍊路資訊

Scala和Java二種方式實戰Spark Streaming開發

Spark基礎:Spark簡介及特點,運作模式,安裝Spark,Driver與Executor,Local模式,Standalone模式,Yarn模式,Mesos模式,WordCount案例,HA配置第1章 Spark概述第2章 Spark運作模式第3章案例實操

Spark實作wordcount

大資料排錯SparkSpark叢集啟動時候，JAVA_HOME is not sethadoop叢集，某台伺服器jps無任何輸出IDEAkafkahadoopspark sqlfile permissionsIDEA本地測試 - OutOfMemoryError: GC overhead limit exceededhdfs負載均衡

spark/scala關于【資源檔案】加載方法概述外部檔案加載方案測試資源檔案打包入jar包中小結