spark streaming 整合kafka 报错 KafkaConsumer is not safe for multi-threaded access

问题描述

spark streaming 使用直连方式读取kafka 数据，使用窗口时出现
java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access
报错信息如图：

代码

object testScala {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName("fs01").setMaster("local[*]")conf.set("spark.streaming.stopGracefullyOnShutdown", "true")conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")conf.set("spark.streaming.kafka.maxRatePerPartition","10")val sc = new SparkContext(conf)sc.setLogLevel("ERROR")val interval = PropertiesUtils.loadProperties("streaming.interval").toLongval ssc:StreamingContext = new StreamingContext(sc, Seconds(5))val kalfa_server_list: String = PropertiesUtils.loadProperties("kafka.broker.list")val kafka_group: String = "group_test_role_1"val kafkaParams = Map[String, Object]("bootstrap.servers" ->  kalfa_server_list, "key.deserializer" -> classOf[StringDeserializer],"value.deserializer" -> classOf[StringDeserializer],"group.id" ->  kafka_group, //消费者组名kafka.groupId"auto.offset.reset" -> "earliest", //earliest可以获取历史数据"enable.auto.commit" -> "false") //如果是true，则这个消费者的偏移量会在后台自动提交val topics = Array("t_2021-09") var offsetRanges: Array[OffsetRange] = Array.empty[OffsetRange]val kafkaStream = KafkaUtils.createDirectStream[String, String](ssc,PreferConsistent, //多数时候采用该方式，在所有可用的executor上均匀分配kafka的主题的所有分区。Subscribe[String, String](topics, kafkaParams))val cacheOper = kafkaStream.transform(rdd=>{offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRangesval check = rdd.map(x => {x.value()})//.cache()check}) cacheOper.window(Seconds(20),Seconds(10)).foreachRDD(rdd=>{rdd.foreach(x=>{println("data:"+x)Thread.sleep(2000L)})kafkaStream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)})ssc.start()ssc.awaitTermination()}
}

解决办法

添加 conf.set(“spark.streaming.kafka.consumer.cache.enabled”, “false”)
这个问题发生的原因是spark 缓存问题，可以查看官网spark streaming整合kafka官网地址

spark streaming 整合kafka 报错 KafkaConsumer is not safe for multi-threaded access相关推荐

【Flink】flink消费kafka报错 KafkaConsumer.assign Ljava/util/List
文章目录 1.概述 1.概述 flink消费kafka上数据时报错: Caused by: java.lang.NoSuchMethodError: org.apache.kafka.clients. ...
【Spark】Spark Stream读取kafka写入kafka报错 AbstractMethodError
1.概述根据这个博客 [Spark]Spark 2.4 Stream 读取kafka 写入kafka 报错如下 Exception in thread "main" java.l ...
spark streaming运行kafka数据源
一.Kafka准备工作 Kafka的安装,请看另外一文,一定要选择和自己电脑上已经安装的scala版本号一致才可以,本教程安装的Spark版本号是1.6.2,scala版本号是2.10,所以,一定要选 ...
【Flink】Flink 写入 kafka 报错 The server disconnected before a response was received
文章目录 1.场景再现 1.1.概述 1.场景再现 1.1.概述 Flink 写入 kafka 报错 The server disconnected before a response was rec ...
Flink读取Kafka报错：KafkaException ByteArrayDeserializer is not an instance Deserializer
1.视界 2.背景做flink读取kafka报错 org.apache.kafka.common.KafkaException: Failed to construct kafka consumer ...
根据官网文档看Spark Streaming对接Kafka的两种方式, 以及如何实现Exactly Once语义
注: 本文算是本人的学习记录, 中间可能有些知识点并不成熟, 不能保证正确性. 只能算是对官网文档作了个翻译和解读, 随时有可能回来更新和纠错上一篇文章讨论了Spark Streaming的WAL( ...
Structured Streaming整合kafka
Structured Streaming整合kafka Spark2.0以后开始推出Structured Streaming,详情参考上文Spark2.0 Structured Streaming.本 ...
Spark Streaming使用Kafka保证数据零丢失
为什么80%的码农都做不了架构师?>>> 源文件放在github,随着理解的深入,不断更新,如有谬误之处,欢迎指正.原文链接https://github.com/jacksu/ ...
spark streaming 消费 kafka入门采坑解决过程
kafka 服务相关的命令 # 开启kafka的服务器 bin/kafka-server-start.sh -daemon config/server.properties & # 创建top ...

spark streaming 整合kafka 报错 KafkaConsumer is not safe for multi-threaded access

问题描述

代码

解决办法

spark streaming 整合kafka 报错 KafkaConsumer is not safe for multi-threaded access相关推荐

最新文章

热门文章