版本: 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

Logo

Kafka开源项目指南提供详尽教程,助开发者掌握其架构、配置和使用,实现高效数据流管理和实时处理。它高性能、可扩展,适合日志收集和实时数据处理,通过持久化保障数据安全,是企业大数据生态系统的核心。

更多推荐