sparkstreaming ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access

116 阅读 0 评论 77 点赞

我是靠谱客的博主拉长大地，这篇文章主要介绍sparkstreaming ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access，现在分享给大家，希望可以做个参考。

最近在测试sparkstreaming的时候发现了一个问题,记录一下
环境 spark 2.x, kafka_0.10.x
示例代码:

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
val ssc: StreamingContext = new StreamingContext(sparkSession.sparkContext,Seconds(5))

    val kafkaBrokers:String = "hadoop01:9092,hadoop02:9092,hadoop03:9092"
    val kafkaTopics: String = "test"

    val kafkaParam = Map(
      "bootstrap.servers" -> kafkaBrokers,//用于初始化链接到集群的地址
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "group1",
      "auto.offset.reset" -> "earliest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    val inputDStream=KafkaUtils.createDirectStream[String,String](ssc, LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String,String](Array(kafkaTopics),kafkaParam))

    val valueDStream: DStream[String] = inputDStream.map(_.value())
    valueDStream.foreachRDD(rdd =>{
      val tRDD: RDD[String] = rdd.filter(_.contains("t"))
      val hRDD: RDD[String] = rdd.filter(_.contains("h"))
      tRDD.union(hRDD).foreach(println)
    })

报错信息.java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access
kafka问题
原因分析:
这里的两个rdd读取的是同一份数据，当执行action时，都会触发两次数据的读操作，（rdd中的一个分区对应着topic中的一个分区，也就是说kafka中的一个分区的数据这里被读取了2次）但是，同一个分区的数据只能被一个consumer消费，所以这里报错。
解决方法：一个可行的解决方案是对rdd进行缓存或者checkpoint，然后要能保证，原始的kafka中的数据，只会被消费一次，然后剩下的数据消费都从缓存中获取数据。
示例代码:

复制代码

1
2
3
4
5
6
7
val valueDStream: DStream[String] = inputDStream.map(_.value()).persist(StorageLevel.DISK_ONLY)
    valueDStream.foreachRDD(rdd =>{
      val tRDD: RDD[String] = rdd.filter(_.contains("t"))
      val hRDD: RDD[String] = rdd.filter(_.contains("h"))
      tRDD.union(hRDD).foreach(println)
    })

参考:java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access
解决KafkaConsumer多线程接入不安全问题（spark streaming 消费kafka）