我正在尝试将 Kafka 集成到我的 Spark 应用程序中,这是我的 POM 文件所需的条目:
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
<version>${spark.stream.kafka.version}</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
<version>${kafka.version}</version>
</dependency>
对应的神器版本有:
<kafka.version>0.10.2.0</kafka.version>
<spark.stream.kafka.version>2.2.0</spark.stream.kafka.version>
我一直在挠头:
Exception in thread "main" java.lang.ClassNotFoundException: Failed to find data source: kafka. Please find packages at http://spark.apache.org/third-party-projects.html
我还尝试为 jar 提供
--jars
参数,但它没有帮助。我在这里想念什么?代码:
private static void startKafkaConsumerStream() {
Dataset<HttpPackage> ds1 = _spark
.readStream()
.format("kafka")
.option("kafka.bootstrap.servers", getProperty("kafka.bootstrap.servers"))
.option("subscribe", HTTP_FED_VO_TOPIC)
.load() // Getting the error here
.as(Encoders.bean(HttpPackage.class));
ds1.foreach((ForeachFunction<HttpPackage>) req ->System.out.print(req));
}
_spark 定义为:
_spark = SparkSession
.builder()
.appName(_properties.getProperty("app.name"))
.config("spark.master", _properties.getProperty("master"))
.config("spark.es.nodes", _properties.getProperty("es.hosts"))
.config("spark.es.port", _properties.getProperty("es.port"))
.config("spark.es.index.auto.create", "true")
.config("es.net.http.auth.user", _properties.getProperty("es.net.http.auth.user"))
.config("es.net.http.auth.pass", _properties.getProperty("es.net.http.auth.pass"))
.getOrCreate();
我的进口是:
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.spark.api.java.function.ForeachFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.SparkSession;
但是,当我按照提到的 here 运行我的代码时并且与包选项有关:
--packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.1.0
有用
最佳答案
Spark 结构化流支持 Apache Kafka 作为流源和接收器,使用外部 kafka-0-10-sql模块。kafka-0-10-sql
模块对使用 spark-submit
提交执行的 Spark 应用程序不可用.该模块是外部的,要使其可用,您应该将其定义为依赖项。
除非你使用 kafka-0-10-sql
Spark 应用程序中的模块特定代码,您不必将模块定义为 dependency
在 pom.xml
.您根本不需要 编译依赖 在模块上,因为没有代码使用模块的代码。您针对接口(interface)进行编码,这也是 Spark SQL 使用起来如此愉快的原因之一(即,它只需要很少的代码就可以拥有相当复杂的分布式应用程序)。spark-submit
但是需要--packages
您报告的命令行选项运行良好。
However when I run my code as mentioned here and which is with the package option:
--packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.1.0
它与
--packages
配合良好的原因是你必须告诉 Spark 基础设施在哪里可以找到 kafka
的定义格式。这导致我们使用 Kafka 运行流式 Spark 应用程序的另一个“问题”(或要求)。您必须指定 运行时依赖 在
spark-sql-kafka
模块。使用
--packages
指定运行时依赖项命令行选项(在您 spark-submit
您的 Spark 应用程序之后下载必要的 jar)或创建所谓的 uber-jar(或 fat-jar)。这就是
pom.xml
来玩(这就是为什么人们用 pom.xml
和模块作为 dependency
提供帮助的原因)。因此,首先,您必须在
pom.xml
中指定依赖项。 .<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql-kafka-0-10_2.11</artifactId>
<version>2.2.0</version>
</dependency>
最后但并非最不重要的一点是,您必须构建一个您在 pom.xml
中配置的 uber-jar。使用 Apache Maven Shade Plugin .使用 Apache Maven Shade 插件 create an Uber JAR这将包括
kafka
的所有“基础设施”格式工作,在 Spark 应用程序 jar 文件中。事实上,Uber JAR 将包含所有必要的运行时依赖项,因此您可以 spark-submit
单独使用 jar (并且没有 --packages
选项或类似选项)。
关于maven - 为什么 spark-submit 找不到 kafka 数据源,除非使用 --packages?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/46001583/