sparkstreaming多consumer消费kafka报错问题
版本: sparkstreaming 2.2kafka 0.10sparkstreaming 集成kafka后(Direct模式),同一个groupid下的多个spark-streaming consumer消费kafka中的数据时,会报错。(如何重现问题:把sparkstreaming集合kafka的程序,运行两次即可)报错如下:该主题下的数据不能访问。问题剖析:假定消费主题为he...
版本: 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
更多推荐
所有评论(0)