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

Posted 2023-03-10

【中文标题】结构化流 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

【发布时间】：2019-11-28 06:30:18

【问题描述】：

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

我下载了结构化流所必需的 spark-streaming-kafka-0-10_2.11.jar (http://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html) 并将其作为“依赖项工件”添加到 Zeppelin 的“spark”解释器中（即还处理 %pyspark 段落）。我重新启动了这个解释器（还有 zeppelin）。

我还在笔记本的第一个段落中加载了 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()

但我得到了错误

执行第 6 行失败：.option("subscribe", "test") \ Traceback （最近一次通话最后）：文件 "/usr/local/analysis/spark/python/lib/pyspark.zip/pyspark/sql/utils.py", 第 63 行，在装饰中 返回 f(*a, **kw) 文件“/usr/local/analysis/spark/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py”, 第 328 行，在 get_return_value 中 format(target_id, ".", name), value) py4j.protocol.Py4JJavaError: 调用 o120.load 时出错。 ： org.apache.spark.sql.AnalysisException：找不到数据源： 卡夫卡。请按照部署部分部署应用程序 《结构化流+Kafka集成指南》。在 org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:652) 在 org.apache.spark.sql.streaming.DataStreamReader.load(DataStreamReader.scala:161) 在 sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:498) 在 py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244) 在 py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357) 在 py4j.Gateway.invoke(Gateway.java:282) 在 py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) 在 py4j.commands.CallCommand.execute(CallCommand.java:79) 在 py4j.GatewayConnection.run(GatewayConnection.java:238) 在 java.lang.Thread.run(Thread.java:748)

在处理上述异常的过程中，又发生了一个异常：

Traceback（最近一次调用最后一次）：文件 “/tmp/zeppelin_pyspark-312826888257172599.py”，第 380 行，在 exec(code, _zcUserQueryNameSpace) File "", line 6, in File "/usr/local/analysis/spark/python/lib/pyspark.zip/pyspark/sql/streaming.py", 第 400 行，负载中 返回 self._df(self._jreader.load()) 文件 "/usr/local/analyse/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py", 第 1257 行，在 调用 答案，self.gateway_client，self.target_id，self.name）文件“/usr/local/analysis/spark/python/lib/pyspark.zip/pyspark/sql/utils.py”， 第 69 行，在装饰中 raise AnalysisException(s.split(': ', 1)[1], stackTrace) pyspark.sql.utils.AnalysisException: '找不到数据源： 卡夫卡。请按照部署部分部署应用程序 《结构化流+Kafka集成指南》。；'

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

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

所以，我还将 spark-streaming-kafka-0-10_2.11.jar 复制到 /usr/local/analyse/spark/jars/ 其他所有的 spark 罐子所在的位置。但是在重新启动（火花和齐柏林飞艇）后，我总是得到同样的错误。

与此同时，我发现我可以在 webbrowser 中查看 spark 的环境变量，并在 "Classpath Entries" 部分中找到 spark-streaming-kafka-0-10_2.11.jar 和源" System Classpath”以及“由用户添加”（似乎是 Zeppelin 解释器部分中的工件）。所以看来我的前两次尝试应该奏效了。

【参考方案1】：

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

对于结构化流，您需要下载包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")

【讨论】：

我把jar文件放到spark的jar目录下。