我正在尝试使用 yarn 与 Kafka 一起运行 spark 流应用程序。我收到以下堆栈跟踪错误-
原因:org.apache.kafka.common.config.ConfigException:缺少没有默认值的必需配置“partition.assignment.strategy”。在 org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:124) 在 org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:48) 在 org.apache.kafka.clients。 org.apache.kafka.clients.consumer.KafkaConsumer.(KafkaConsumer.java:380) 上的 consumer.ConsumerConfig.(ConsumerConfig.java:194) org.apache.kafka.clients.consumer.KafkaConsumer.(KafkaConsumer.java:363) ) 在 org.apache.spark.streaming.kafka010.CachedKafkaConsumer.(CachedKafkaConsumer.scala:45) 在 org.apache.spark.streaming 的 org.apache.kafka.clients.consumer.KafkaConsumer.(KafkaConsumer.java:350)。 kafka010.CachedKafkaConsumer$.get(CachedKafkaConsumer.scala:194) 在 org.
这是我如何使用火花流创建 KafkaStream 的代码片段-
val ssc = new StreamingContext(sc, Seconds(60))
val kafkaParams = Map[String, Object](
"bootstrap.servers" -> "*boorstrap_url:port*",
"security.protocol" -> "SASL_PLAINTEXT",
"sasl.kerberos.service.name" -> "kafka",
"key.deserializer" -> classOf[StringDeserializer],
"value.deserializer" -> classOf[StringDeserializer],
"group.id" -> "annotation-test",
//Tried commenting and uncommenting this property
//"partition.assignment.strategy"->"org.apache.kafka.clients.consumer.RangeAssignor",
"auto.offset.reset" -> "earliest",
"enable.auto.commit" -> (false: java.lang.Boolean))
val topics = Array("*topic-name*")
val kafkaStream = KafkaUtils.createDirectStream[String, String](
ssc,
PreferConsistent,
Subscribe[String, String](topics, kafkaParams))
val valueKafka = kafkaStream.map(record => record.value())
我浏览了以下帖子-
据此,我已将我的 fat jar 中的 kafka util jar 从默认从 spark-stream-kafka-jar 打包的 0.10.1.0 更新为0.10.2.0版本作为瞬态依赖项。当我通过将master设置为本地来在单个节点上运行它时,我的工作也可以正常工作。我正在运行 spark 2.3.1 版本。