交流QQ: 824203453
 

版本： sparkstreaming 2.2      kafka  0.10

sparkstreaming 整合kafka後（Direct模式），同一個groupid下的多個spark-streaming consumer消費kafka中的資料時，會報錯。（如何重現問題：把sparkstreaming集合kafka的程式，執行兩次即可）

報錯如下：

該主題下的資料不能訪問。

問題剖析：

假定消費主題為helloTopic8 , 該主題設定的分割槽數量為3個。

當使用kafka的javaAPI消費資料時，如果啟動同一個組下的多個consumer，程式不會報錯，但是真正能消費到資料的consumer數量，只能和消費的主題的partition的個數一致（這裡為3個）。其他的consumer不能消費到資料。

而當spark streaming 程式整合kafka後，DStream中的rdd的分割槽數和消費的helloTopic8的partition數量一致（也就是3個），所以執行spark streaming程式，也就相當於啟動了3個task來讀取kafka中helloTopic8的資料。

當重複再啟動一個程式時，會重新消費到helloTopic8中的3歌分割槽的數量，導致原來的程式讀取不到相應的資料，隨即報錯。

示例程式碼：

    val conf = new SparkConf()
    .setAppName(this.getClass.getSimpleName)
    .setMaster("local[*]")
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))

    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hdp-02:9092,hdp-03:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer], 
      "group.id" -> "group_hello",
      "auto.offset.reset" -> "earliest" // 最早的
//      "enable.auto.commit" -> (false: java.lang.Boolean)
     ) // 是否自動提交offset

    // 通過呼叫KafkaUtils API 來建立一個DStream
    val topics = Array("helloTopic8")
    val directStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(ssc,
      LocationStrategies.PreferConsistent,
      // 訂閱主題 注意需要給定訊息的型別
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
    )
    directStream.map(_.value()).map((_,1)).reduceByKey(_+_).print()
    // 啟動  阻塞
    ssc.start()
    ssc.awaitTermination()
 

該問題 類似於下面這個問題：

Caused by: java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access

可參考：https://blog.csdn.net/qq_21439395/article/details/80412688

交流QQ: 824203453