结构化流式传输 Kafka 2.1->Zeppelin 0.8->Spark 2.4：spark 不使用 jar

Question

我有一个 Kafka 2.1 消息代理，想在 Spark 2.4 中对消息数据进行一些处理。我想使用 Zeppelin 0.8.1 notebooks 进行快速原型制作。

我下载了结构化流媒体 (http://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html) 所必需的 spark-streaming-kafka-0-10_2.11.jar 并将其作为 "Dependencies-artifact" 添加到 "spark"-interpreter Zeppelin 的（也处理 %pyspark 段落）。我重新启动了这个解释器（还有飞艇）。

我还在第一个笔记本段落中加载了jar（我首先认为这应该不是必需的......）：

%dep z.load("/usr/local/analyse/jar/spark-streaming-kafka-0-10_2.11.jar")
res0: org.apache.zeppelin.dep.Dependency = org.apache.zeppelin.dep.Dependency@2b65d5

所以，我没有收到任何错误，所以加载似乎有效。现在，我想做测试，kafka 服务器使用这个端口在同一台机器上运行，还有一个主题 "test":

%pyspark
# Subscribe to a topic
df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "localhost:9092") \
  .option("subscribe", "test") \
  .load()

但是我得到了错误

Fail to execute line 6: .option("subscribe", "test") \ Traceback (most recent call last): File "/usr/local/analyse/spark/python/lib/pyspark.zip/pyspark/sql/utils.py", line 63, in deco return f(*a, **kw) File "/usr/local/analyse/spark/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py", line 328, in get_return_value format(target_id, ".", name), value) py4j.protocol.Py4JJavaError: An error occurred while calling o120.load. : org.apache.spark.sql.AnalysisException: Failed to find data source: kafka. Please deploy the application as per the deployment section of "Structured Streaming + Kafka Integration Guide".; at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:652) at org.apache.spark.sql.streaming.DataStreamReader.load(DataStreamReader.scala:161) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244) at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357) at py4j.Gateway.invoke(Gateway.java:282) at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) at py4j.commands.CallCommand.execute(CallCommand.java:79) at py4j.GatewayConnection.run(GatewayConnection.java:238) at java.lang.Thread.run(Thread.java:748)

During handling of the above exception, another exception occurred:

Traceback (most recent call last): File "/tmp/zeppelin_pyspark-312826888257172599.py", line 380, in exec(code, _zcUserQueryNameSpace) File "", line 6, in File "/usr/local/analyse/spark/python/lib/pyspark.zip/pyspark/sql/streaming.py", line 400, in load return self._df(self._jreader.load()) File "/usr/local/analyse/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py", line 1257, in call answer, self.gateway_client, self.target_id, self.name) File "/usr/local/analyse/spark/python/lib/pyspark.zip/pyspark/sql/utils.py", line 69, in deco raise AnalysisException(s.split(': ', 1)[1], stackTrace) pyspark.sql.utils.AnalysisException: 'Failed to find data source: kafka. Please deploy the application as per the deployment section of "Structured Streaming + Kafka Integration Guide".;'

我想知道至少其中一项调整（解释器配置或直接加载）应该有效。

我还在控制台上尝试了 spark-submit --jar /usr/local/analyse/jar/spark-streaming-kafka-0-10_2.11.jar 但这似乎只有在我提交程序时才有效。

因此，我也将 spark-streaming-kafka-0-10_2.11.jar 复制到 /usr/local/analyse/spark/jars/ 其他所有 spark 罐所在的位置。但是在重新启动（spark 和 zeppelin）之后，我总是得到同样的错误。

与此同时，我发现我可以在网络浏览器中查看 spark 的环境变量，并且在 spark-streaming-kafka-0-10_2.11.jar 部分找到了源代码 "Classpath Entries" "System Classpath" 和 "Added By User" （似乎是 Zeppelin 解释器部分的神器）。所以看来我的前两次尝试应该奏效了。

Answer 1

第一个问题是您已经下载了用于 spark streaming 的包，但尝试创建一个结构化流对象（readstream()）。请记住，Spark Streaming 和 Spark Structured Streaming 是两种不同的东西，需要区别对待。

对于结构化流式传输，您需要下载包 spark-sql-kafka-0-10_2.11 and its dependencies kafka-clients, slf4j-api, snappy-java, lz4-java and unused。你的依赖部分应该像这样加载所有需要的包：

z.load("/tmp/spark-sql-kafka-0-10_2.11-2.4.0.jar")
z.load("/tmp/kafka-clients-2.0.0.jar")
z.load("/tmp/lz4-java-1.4.0.jar")
z.load("/tmp/snappy-java-1.1.7.1.jar")
z.load("/tmp/unused-1.0.0.jar")
z.load("/tmp/slf4j-api-1.7.16.jar")

结构化流式传输 Kafka 2.1->Zeppelin 0.8->Spark 2.4：spark 不使用 jar

structured streaming Kafka 2.1->Zeppelin 0.8->Spark 2.4: spark does not use jar

python

apache-kafka

apache-spark

pyspark

apache-zeppelin