1
Spark 스트리밍 응용 프로그램을 사용하고 있습니다. 응용 프로그램은 직접 스트림을 사용하여 Kafka 항목 (200 개의 파티션)에서 메시지를 읽습니다. 때때로 응용 프로그램이 ConcurrentModificationException->Spark Streaming kafka 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)
내 스파크 클러스터는 두 개의 노드가 발생합니다. 스파크 버전은 2.1입니다. 응용 프로그램은 두 개의 실행 프로그램을 실행합니다. 예외 및 카프카 소비자 코드에서 확인할 수 있듯이 동일한 kakfa 소비자가 두 개의 스레드에 의해 사용되고있는 것으로 보입니다. 두 스레드가 동일한 수신기에 액세스하는 방법을 알지 못합니다. 이상적으로 각 집행자는 할당 된 모든 파티션에 대한 메시지를 읽어야하는 단일 스레드에 의해 독점적 인 카프카 수신자 서비스를 가져야합니다. 내 경우> kafka-에서
JavaInputDStream<ConsumerRecord<String, String>> consumerRecords = KafkaUtils.createDirectStream(
jssc,
LocationStrategies.PreferConsistent(),
ConsumerStrategies.<String, String>Subscribe(topics, kafkaParams));
사용중인 코드를 추가하십시오. –
은 kafka에서 읽은 코드 스 니펫을 추가했습니다. – scorpio
단일 실행 프로그램에서 다중 작업을 실행 했습니까? 'useConsumerCache'를 false로 설정하십시오. 'KafkaConsumer'는 스레드로부터 안전하지 않습니다. 즉, 여러 스레드에서 사용할 수 없습니다. – amethystic