Spark Streaming kafka concurrentModificationException答案

【问题标题】：Spark Streaming kafka concurrentModificationExceptionSpark Streaming kafka concurrentModificationException
【发布时间】：2018-05-17 01:54:23
【问题描述】：

我正在使用 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$1.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 集群有两个节点。 Spark 版本是 2.1。该应用程序运行两个执行程序。从我从异常和 kafka 消费者代码中可以看出，似乎两个线程正在使用同一个 kakfa 消费者。我不知道两个线程是如何访问同一个接收器的。理想情况下，每个执行程序都应该有一个由单个线程提供的专有 kafka 接收器服务，该线程必须读取所有分配的分区的消息。从kafka读取的代码sn-p->

JavaInputDStream<ConsumerRecord<String, String>> consumerRecords = KafkaUtils.createDirectStream(
jssc,
LocationStrategies.PreferConsistent(),
ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams));

【问题讨论】：

添加您正在使用的代码。
添加了我从 kafka 读取的代码 sn-p。
你在单个执行器中运行多任务吗？尝试将 useConsumerCache 设置为 false。 KafkaConsumer 不是线程安全的，即不能在多个线程中使用。
@amethystic：我正在为每项工作提交火花。我仔细检查了所有任务都有各自的执行者（确切地说是一个驱动程序和两个执行者）。
还有这方面的信息吗？我仍然面临这个问题。

标签： apache-spark apache-kafka streaming spark-streaming-kafka

【解决方案1】：

就我而言，问题与 kafka 消费者缓存大小有关。我将大小（默认值：每个执行程序 64 个）更改为每个执行程序 200 个（由于 200 个分区，200 个并行消费者）。我不得不升级到 Spark 2.2，因为 Spark 2.1 中没有更改大小的选项。

spark.streaming.kafka.consumer.cache.maxCapacity=200

【讨论】：

如果我有 200 个分区并且我有 2 个执行器。我应该将 spark.streaming.kafka.consumer.cache.maxCapacity 增加到 100 吗？因为 2 个执行者*100 == 200？
该设置针对每个执行程序 VM。由于您有 200 个分区，因此每个 VM 的分区将为 100。因此，您只需将 maxCapacity 设置配置为 100。