Spark Streaming kafka concurrentModificationException
Spark Streaming kafka concurrentModificationException
我正在使用 Spark 流应用程序。应用程序使用直接流从 Kafka 主题（具有 200 个分区）读取消息。偶尔应用程序会抛出 ConcurrentModificationException->
java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access
at org.apache.kafka.clients.consumer.KafkaConsumer.acquire(KafkaConsumer.java:1431)
at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:1361)
at org.apache.spark.streaming.kafka010.CachedKafkaConsumer$$anon.removeEldestEntry(CachedKafkaConsumer.scala:128)
at java.util.LinkedHashMap.afterNodeInsertion(LinkedHashMap.java:299)
at java.util.HashMap.putVal(HashMap.java:663)
at java.util.HashMap.put(HashMap.java:611)
at org.apache.spark.streaming.kafka010.CachedKafkaConsumer$.get(CachedKafkaConsumer.scala:158)
at org.apache.spark.streaming.kafka010.KafkaRDD$KafkaRDDIterator.<init>(KafkaRDD.scala:211)
at org.apache.spark.streaming.kafka010.KafkaRDD.compute(KafkaRDD.scala:186)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
at org.apache.spark.scheduler.Task.run(Task.scala:99)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:282)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
我的 spark 集群有两个节点。星火版本是2.1。该应用程序运行两个执行程序。从我从异常和 kafka 消费者代码中可以看出，两个线程似乎正在使用同一个 kakfa 消费者。我不知道为什么两个线程会访问同一个接收器。理想情况下，每个执行者都应该有一个单独的 kafka 接收器服务，该线程必须读取所有分配分区的消息。
从 kafka->
读取的代码片段
JavaInputDStream<ConsumerRecord<String, String>> consumerRecords = KafkaUtils.createDirectStream(
jssc,
LocationStrategies.PreferConsistent(),
ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams));
就我而言，问题与 kafka 消费者缓存大小有关。我将大小（默认值：每个执行者 64 个）更改为每个执行者 200 个（由于 200 个分区，200 个并行消费者）。我不得不升级到 Spark 2.2，因为更改大小的选项在 Spark 2.1 中不可用。
spark.streaming.kafka.consumer.cache.maxCapacity=200
我正在使用 Spark 流应用程序。应用程序使用直接流从 Kafka 主题（具有 200 个分区）读取消息。偶尔应用程序会抛出 ConcurrentModificationException->
java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access
at org.apache.kafka.clients.consumer.KafkaConsumer.acquire(KafkaConsumer.java:1431)
at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:1361)
at org.apache.spark.streaming.kafka010.CachedKafkaConsumer$$anon.removeEldestEntry(CachedKafkaConsumer.scala:128)
at java.util.LinkedHashMap.afterNodeInsertion(LinkedHashMap.java:299)
at java.util.HashMap.putVal(HashMap.java:663)
at java.util.HashMap.put(HashMap.java:611)
at org.apache.spark.streaming.kafka010.CachedKafkaConsumer$.get(CachedKafkaConsumer.scala:158)
at org.apache.spark.streaming.kafka010.KafkaRDD$KafkaRDDIterator.<init>(KafkaRDD.scala:211)
at org.apache.spark.streaming.kafka010.KafkaRDD.compute(KafkaRDD.scala:186)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
at org.apache.spark.scheduler.Task.run(Task.scala:99)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:282)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
我的 spark 集群有两个节点。星火版本是2.1。该应用程序运行两个执行程序。从我从异常和 kafka 消费者代码中可以看出，两个线程似乎正在使用同一个 kakfa 消费者。我不知道为什么两个线程会访问同一个接收器。理想情况下，每个执行者都应该有一个单独的 kafka 接收器服务，该线程必须读取所有分配分区的消息。 从 kafka->
读取的代码片段JavaInputDStream<ConsumerRecord<String, String>> consumerRecords = KafkaUtils.createDirectStream(
jssc,
LocationStrategies.PreferConsistent(),
ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams));
就我而言，问题与 kafka 消费者缓存大小有关。我将大小（默认值：每个执行者 64 个）更改为每个执行者 200 个（由于 200 个分区，200 个并行消费者）。我不得不升级到 Spark 2.2，因为更改大小的选项在 Spark 2.1 中不可用。
spark.streaming.kafka.consumer.cache.maxCapacity=200