【发布时间】:2019-11-04 08:16:24
【问题描述】:
我正在尝试使用 spark 阅读 kafka,但我猜遇到了一些与库相关的问题。
我正在向 kafka 主题推送一些事件,我可以通过 kafka 控制台消费者阅读但无法通过 spark 阅读。我正在使用 spark-sql-kafka 库,该项目是用 maven 编写的。 Scala 版本是 2.11.12,spark 版本是 2.4.3。
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.11</artifactId>
<version>2.4.3</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.11</artifactId>
<version>2.4.3</version>
<scope>provided</scope>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.spark/spark-sql-kafka-0-10 -->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql-kafka-0-10_2.11</artifactId>
<version>2.4.3</version>
<scope>provided</scope>
</dependency>
我的java代码如下:-
SparkSession spark = SparkSession.builder()
.appName("kafka-tutorials")
.master("local[*]")
.getOrCreate();
Dataset<Row> rows = spark.readStream().
format("kafka").option("kafka.bootstrap.servers", "localhost:9092")
.option("subscribe", "meetup-trending-topics")
.option("startingOffsets", "latest")
.load();
rows.writeStream()
.outputMode("append")
.format("console")
.start();
spark.streams().awaitAnyTermination();
spark.stop();
以下是我收到的错误消息:-
线程“main”org.apache.spark.sql.AnalysisException 中的异常:找不到数据源:kafka。请按照《Structured Streaming + Kafka Integration Guide》部署部分部署应用; 在 org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:652) 在 org.apache.spark.sql.streaming.DataStreamReader.load(DataStreamReader.scala:161)
解决方案:- 1)创建 uber jar 或 ii)--packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.3 我之前在 mainclass 之后给出了 --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.3 选项。
【问题讨论】:
-
您使用了不兼容的工件版本(一些使用 Scala 2.11 编译,其他使用 2.12)
标签: java maven apache-spark apache-kafka spark-streaming