sparkstreaming多consumer消费kafka报错问题

版本： 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 来创建一个DStreamval 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

楼主建议遵循如下原则：

即同一个partition内的消息只能被同一个组中的一个consumer消费。当消费者数量多于partition的数量时，多余的消费者空闲。
也就是说如果只有一个partition你在同一组启动多少个consumer都没用，partition的数量决定了此topic在同一组中被可被均衡的程度，例如partition=4，则可在同一组中被最多4个consumer均衡消费。

对于楼主如果三个分区，建议三个消费者，不建议启动第四个。