Spark Streaming讀取Kafka數據的兩種方式

Kafka在0.8和0.10之間引入了一種新的消費者API,因此,Spark Streaming與Kafka集成,有兩種包可以選擇: spark-streaming-kafka-0-8與spark-streaming-kafka-0-10。在使用時應注意以下幾點:

1. spark-streaming-kafka-0-8兼容Kafka 0.8.2.1及以后的版本, 從Spark 2.3.0開始，對Kafka 0.8支持已被標記為過時。

2. spark-streaming-kafka-0-10兼容Kafka 0.10.0及以后的版本, 從Spark 2.3.0開始, 此API是穩定版。

3. 如果Kafka版本大於等於0.10.0，且Spark版本大於等於Spark 2.3.0，應使用spark-streaming-kafka-0-10。

本文總結spark-streaming-kafka-0-8中兩種讀取Kafka數據的方式:createStream、createDirectStream。

基於Receiver方式

POM依賴

<dependencies>
     <!--spark-streaming-->
     <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-streaming_2.11</artifactId>
         <version>2.2.2</version>
     </dependency>

     <!--spark-streaming-kafka-plugin-->
     <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
         <version>2.2.2</version>
     </dependency>
 </dependencies>

示例一

// 1、Kafka配置
 // 配置zookeeper集群、消費者組
 val kafkaParams = Map(
   "zookeeper.connect" -> "localhost:2181",
   "group.id" -> groupID)

 // 2、topic_name與numThreads的映射
 // topic有幾個partition,就寫幾個numThreads。
 // 每個partition對應一個單獨線程從kafka取數據到Spark Streaming
 val topics = Map(topicName -> numThreads)

 // 3、ReceiverInputDStream
 // 注意:應先import kafka.serializer.StringDecoder再import org.apache.spark.streaming._
 val kafkaStream= KafkaUtils.createStream[String, String, StringDecoder, StringDecoder](
   ssc,
   kafkaParams,
   topics,
   StorageLevel.MEMORY_AND_DISK_SER_2)

示例二

// 1、topic_name與numThreads的映射
 // topic有幾個partition,就寫幾個numThreads。
 // 每個partition對應一個單獨線程從kafka取數據到Spark Streaming
 val topics = Map(topicName -> numThreads)

 // 2、ReceiverInputDStream
 // 底層先根據zkQuorum、groupId 構造kafkaParams,
 // 然后再調用createStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics, storageLevel)
 val kafkaStream=KafkaUtils.createStream(
   ssc=ssc,
   zkQuorum="localhost:2181",
   groupId = groupID,
   topics,
   StorageLevel.MEMORY_AND_DISK_SER_2
 )

特點

1. 需要使用單獨的Receiver線程來異步獲取Kafka數據。

2. Receiver底層實現中使用了Kafka高級消費者API,因此,不需要自己管理Offset,只需指定Zookeeper和消費者組GroupID,系統便會自行管理。

3. 執行過程: Spark Streaming啟動時，會在Executor中同時啟動Receiver異步線程用於從Kafka持續獲取數據，獲取的數據先存儲在Receiver中(存儲方式由StorageLevel決定)，后續，當Batch Job觸發后，這些數據會被轉移到剩下的Executor中被處理。處理完畢后，Receiver會自動更新Zookeeper中的Offset。

4. 默認情況下,程序失敗或Executor宕掉后可能會丟失數據，為避免數據丟失，可啟用預寫日志(Write Ahead Log,WAL)。將Receiver收到的數據再備份一份到更可靠的系統如HDFS分布式文件中，以冗余的數據來換取數據不丟失。

5. 生產下，為保證數據完全不丟失，一般需要啟用WAL。啟用WAL，在數據量較大，網絡不好情況下，會嚴重降低性能。

---

基於Direct(No Receiver)方式

POM依賴

<dependencies>
     <!--spark-streaming-->
     <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-streaming_2.11</artifactId>
         <version>2.3.1</version>
     </dependency>

     <!--spark-streaming-kafka-plugin-->
     <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
         <version>2.3.1</version>
     </dependency>
 </dependencies>

示例一

// 1、Kafka配置
 // auto.offset.reset=latest 無提交的offset時,從最新的開始消費
 // enable.auto.commit=false 禁用后台自動提交offset,自己手動管理
 val kafkaParams = Map[String, Object](
   "bootstrap.servers" -> "localhost:9092",
   "key.deserializer" -> classOf[StringDeserializer],
   "value.deserializer" -> classOf[StringDeserializer],
   "auto.offset.reset" -> "latest",
   "enable.auto.commit" -> (false: java.lang.Boolean),
   "group.id" -> groupID)

 // 2、DirectKafkaInputDStream
 // LocationStrategies:本地策略。為提升性能,可指定Kafka Topic Partition的消費者所在的Executor。
 // LocationStrategies.PreferConsistent:一致性策略。一般情況下用這個策略就OK。將分區盡可能分配給所有可用Executor。
 // LocationStrategies.PreferBrokers:特殊情況,如果Executor和Kafka Broker在同一主機,則可使用此策略。
 // LocationStrategies.PreferFixed:特殊情況,當Kafka Topic Partition負荷傾斜,可用此策略,手動指定Executor來消費特定的Partition.
 // ConsumerStrategies:消費策略。
 // ConsumerStrategies.Subscribe/SubscribePattern:可訂閱一類Topic,且當新Topic加入時，會自動訂閱。一般情況下，用這個就OK。
 // ConsumerStrategies.Assign:可指定要消費的Topic-Partition,以及從指定Offset開始消費。
 val kafkaStream=KafkaUtils.createDirectStream[String,String](
   ssc,
   LocationStrategies.PreferConsistent,
   ConsumerStrategies.Subscribe[String,String](List(topicName),kafkaParams)
 )

示例二

// 1、Kafka配置
 // auto.offset.reset=latest 無提交的offset時,從最新的開始消費
 // enable.auto.commit=false 禁用后台自動提交offset,自己手動管理
 val kafkaParams = Map[String, Object](
   "bootstrap.servers" -> "localhost:9092",
   "key.deserializer" -> classOf[StringDeserializer],
   "value.deserializer" -> classOf[StringDeserializer],
   "auto.offset.reset" -> "latest",
   "enable.auto.commit" -> (false: java.lang.Boolean),
   "group.id" -> groupID)

 // 2、DirectKafkaInputDStream
 // LocationStrategies.PreferConsistent:一致性策略。
 // ConsumerStrategies.Assign:從指定Topic-Partition的Offset開始消費。
 val initOffset=Map(new TopicPartition(topicName,0)->10L)
 val kafkaStream=KafkaUtils.createDirectStream[String,String](
   ssc,
   LocationStrategies.PreferConsistent,
   ConsumerStrategies.Assign[String,String](initOffset.keys,kafkaParams,initOffset)
 )

---

特點

1. 不需要使用單獨的Receiver線程從Kafka獲取數據。

2. 使用Kafka簡單消費者API,不需要ZooKeeper參與，直接從Kafka Broker獲取數據。

3. 執行過程:Spark Streaming Batch Job觸發時，Driver端確定要讀取的Topic-Partition的OffsetRange，然后由Executor並行從Kafka各Partition讀取數據並計算。

4. 為保證整個應用EOS， Offset管理一般需要借助外部存儲實現。如Mysql、HBase等。

5. 由於不需要WAL，且Spark Streaming會創建和Kafka Topic Partition一樣多的RDD Partition,且一一對應，這樣,就可以並行讀取，大大提高了性能。

6. Spark Streaming應用啟動后，自己通過內部currentOffsets變量跟蹤Offset，避免了基於Receiver的方式中Spark Streaming和Zookeeper中的Offset不一致問題。