?spark讀取kafka數(shù)據(jù)

// Create DataFrame representing the stream of input lines from kafkaval lines = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "kafka1:9092,kafka2:9092,kafka3:9092").option("subscribe", "log_active").load()

錯(cuò)誤信息：

21/02/01 10:39:50 WARN consumer.ConsumerConfig: The configuration max.poll.records = 1 was supplied but isn't a known config. 21/02/01 10:39:50 INFO utils.AppInfoParser: Kafka version : 0.9.0-kafka-2.0.2 21/02/01 10:39:50 INFO utils.AppInfoParser: Kafka commitId : unknown 21/02/01 10:39:50 ERROR streaming.StreamExecution: Query [id = 3a0fd490-4f78-4d4f-ac33-a245b04e363f, runId = 2c5f1322-2c8e-4e5a-b992-5b859cb0bdd6] terminated with error java.lang.NoSuchMethodError: org.apache.kafka.clients.consumer.KafkaConsumer.subscribe(Ljava/util/Collection;)Vat org.apache.spark.sql.kafka010.SubscribeStrategy.createConsumer(ConsumerStrategy.scala:63)at org.apache.spark.sql.kafka010.KafkaOffsetReader.createConsumer(KafkaOffsetReader.scala:297)at org.apache.spark.sql.kafka010.KafkaOffsetReader.<init>(KafkaOffsetReader.scala:78)at org.apache.spark.sql.kafka010.KafkaSourceProvider.createSource(KafkaSourceProvider.scala:88)at org.apache.spark.sql.execution.datasources.DataSource.createSource(DataSource.scala:243)at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$2$$anonfun$applyOrElse$1.apply(StreamExecution.scala:158)at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$2$$anonfun$applyOrElse$1.apply(StreamExecution.scala:155)at scala.collection.mutable.MapLike$class.getOrElseUpdate(MapLike.scala:194)at scala.collection.mutable.AbstractMap.getOrElseUpdate(Map.scala:80)at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$2.applyOrElse(StreamExecution.scala:155)at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$2.applyOrElse(StreamExecution.scala:153)at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$2.apply(TreeNode.scala:267)at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$2.apply(TreeNode.scala:267)at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:70)at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:266)at org.apache.spark.sql.catalyst.trees.TreeNode.transform(TreeNode.scala:256)at org.apache.spark.sql.execution.streaming.StreamExecution.logicalPlan$lzycompute(StreamExecution.scala:153)at org.apache.spark.sql.execution.streaming.StreamExecution.logicalPlan(StreamExecution.scala:147)at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runBatches(StreamExecution.scala:276)at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:206) Exception in thread "stream execution thread for [id = 3a0fd490-4f78-4d4f-ac33-a245b04e363f, runId = 2c5f1322-2c8e-4e5a-b992-5b859cb0bdd6]" java.lang.NoSuchMethodError: org.apache.kafka.clients.consumer.KafkaConsumer.subscribe(Ljava/util/Collection;)Vat org.apache.spark.sql.kafka010.SubscribeStrategy.createConsumer(ConsumerStrategy.scala:63)at org.apache.spark.sql.kafka010.KafkaOffsetReader.createConsumer(KafkaOffsetReader.scala:297)at org.apache.spark.sql.kafka010.KafkaOffsetReader.<init>(KafkaOffsetReader.scala:78)at org.apache.spark.sql.kafka010.KafkaSourceProvider.createSource(KafkaSourceProvider.scala:88)at org.apache.spark.sql.execution.datasources.DataSource.createSource(DataSource.scala:243)at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$2$$anonfun$applyOrElse$1.apply(StreamExecution.scala:158)at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$2$$anonfun$applyOrElse$1.apply(StreamExecution.scala:155)at scala.collection.mutable.MapLike$class.getOrElseUpdate(MapLike.scala:194)at scala.collection.mutable.AbstractMap.getOrElseUpdate(Map.scala:80)at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$2.applyOrElse(StreamExecution.scala:155)at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$2.applyOrElse(StreamExecution.scala:153)at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$2.apply(TreeNode.scala:267)at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$2.apply(TreeNode.scala:267)at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:70)at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:266)at org.apache.spark.sql.catalyst.trees.TreeNode.transform(TreeNode.scala:256)at org.apache.spark.sql.execution.streaming.StreamExecution.logicalPlan$lzycompute(StreamExecution.scala:153)at org.apache.spark.sql.execution.streaming.StreamExecution.logicalPlan(StreamExecution.scala:147)at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runBatches(StreamExecution.scala:276)at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:206) Exception in thread "main" org.apache.spark.sql.streaming.StreamingQueryException: org.apache.kafka.clients.consumer.KafkaConsumer.subscribe(Ljava/util/Collection;)V

這個(gè)在官方文檔中有介紹。地址如下：https://www.cloudera.com/documentation/spark2/latest/topics/spark2_kafka.html#running_jobs

方案一：錯(cuò)誤信息中可以看出kafka的版本：Kafka version : 0.9.0-kafka-2.0.2，而我在pom.xml中應(yīng)用的jar是0.10，因此導(dǎo)致包不一致。

# Set the environment variable for the duration of your shell session: export SPARK_KAFKA_VERSION=0.10 spark-submit arguments# Or:# Set the environment variable for the duration of a single command: SPARK_KAFKA_VERSION=0.10 spark-submit arguments

方法二：參照https://docs.cloudera.com/documentation/spark2/latest/topics/spark2_kafka.html#running_jobs

?

總結(jié)

以上是生活随笔為你收集整理的spark2+kafka报错:java.lang.NoSuchMethodError:org.apache.kafka.clients.consumer.KafkaConsumer.subscribe的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。