python - 结构化流 Kafka 2.1->Zeppelin 0.8->Spark 2.4 : spark does not use jar

Question

我有一个 Kafka 2.1 消息代理，想对 Spark 2.4 中的消息数据进行一些处理。我想使用 Zeppelin 0.8.1 笔记本进行快速原型(prototype)制作。

我下载了结构化流媒体 (http://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html) 所必需的 spark-streaming-kafka-0-10_2.11.jar，并将其作为“Dependencies-artifact”添加到 Zeppelin 的“spark”解释器中(即还处理 %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 服务器使用这个端口在同一台机器上运行，还有一个主题“测试”:

%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 jar 所在的位置。但是在重新启动(spark 和 zeppelin)之后，我总是得到同样的错误。

与此同时，我发现我可以在网络浏览器中查看 spark 的环境变量，并且在“类路径条目”部分找到了 spark-streaming-kafka-0-10_2.11.jar 和源“系统类路径”以及“用户添加”(似乎是 Zeppelin 解释器部分中的神器)。所以看来我的前两次尝试应该奏效了。

Answer 1

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

对于结构化流，您需要下载包 spark-sql-kafka-0-10_2.11及其依赖项 kafka-clients , slf4j-api , snappy-java , lz4-java和 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")