Spark streaming和kafka缺少没有默认值的必需配置“partition.assignment.strategy”

Question

我正在尝试使用纱线与Kafka一起运行spark streaming应用程序。我得到以下堆栈跟踪错误-

造成原因：org.apache.kafka.common.config.配置异常：缺少所需的配置partition.assignment.strategy没有默认值。在org.apache.kafka.common.config.配置ef.parse（配置ef.java:124）在org.apache.kafka.common.config.AbstractConfig。（AbstractConfig.java:48）org.apache.kafka.clients.consumer.消费者配置。消费者onfig.java:194）在org.apache.kafka.clients.consumer.Kafka消费者。Kafkaonsumer.java:380）在org.apache.kafka.clients.consumer.Kafka消费者。Kafkaonsumer.java:363）org.apache.kafka.clients.consumer.Kafka消费者。KafkaConsumer.java:350）org.apache.spark.streaming.kafka010。CachedKafka消费者。CachedKafkaConsumer.scala:45）org.apache.spark.streaming.kafka010。消费者$. get（CachedKafkaConsumer.scala:194）在org.apache.spark.streaming.kafka010。KafkaRDDIterator。kafkaRDD. scala： 252）在org. apache. Spark. stream. kafka010。计算（KafkaRDD. scala： 212）在org. apache. spark. rdd。RDD. computeOrReadCheckpoint（RDD. scala： 324）at org. apache. spark. rdd。rdd. iterator（rdd. scala： 288）at org. apache. spark. rdd。scala： 49）在org. apache. spark. rdd。RDD. computeOrReadCheckpoint（RDD. scala： 324）at org. apache. spark. rdd。rdd. iterator（rdd. scala： 288）at org. apache. spark.调度器。运行任务（ResultTasks. scala： 87）在org. apache. Spark.调度器。在org. apache. spak. exitor上运行（Task. scala： 109）。执行器$TaskRunner. run（Executor. scala： 345）

下面是我如何使用spark stream创建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" -> "annotathtml" target="_blank">ion-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())

我已经看过下面的帖子了-

1. https://issues.apache.org/jira/browse/KAFKA-4547

根据这一点，我已经将fat jar中的kafka util jar从spark stream kafka jar默认打包的0.10.1.0版本更新为0.10.2.0版本，作为临时依赖项。当我在单个节点上运行它时，通过将master设置为local，我的工作也可以正常工作。我正在运行spark 2.3.1版本。

Answer 1

添加kafka客户端-*。jar到您的spark jar文件夹<代码>Kafka客户端-*。jar位于kafka-*/lib目录中。