我有一个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文件(起初我认为这不是必要的...)：
但我得到了错误。
执行第6行失败：.option("subscribe", "test") \ Traceback （最近的调用最先）：文件 "/usr/local/analyse/spark/python/lib/pyspark.zip/pyspark/sql/utils.py"， 第63行，在装饰器中： 返回f（* a，** kw）文件“/usr/local/analyse/spark/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py”的行328中，获取返回值： 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（most recent call last）：文件 "/tmp/zeppelin_pyspark-312826888257172599.py"，第380行， exec（code，_zcUserQueryNameSpace）文件“”，第6行，在文件 "/usr/local/analyse/spark/python/lib/pyspark.zip/pyspark/sql/streaming.py"的第400行处加载： return self._df（self._jreader.load（））文件“/usr/local/analyse/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py”的第1257行， 调用 answer，self.gateway_client，self.target_id，self.name）文件“/usr/local/analyse/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/analyse/jar/spark-streaming-kafka-0-10_2.11.jar，但似乎只有在提交程序时才有效。
所以，我也将spark-streaming-kafka-0-10_2.11.jar复制到了/usr/local/analyse/spark/jars/中，那里是spark的所有其他jars。但是，在重新启动（spark和zeppelin）后，我仍然得到相同的错误。
与此同时，我发现我可以在Web浏览器中查看spark的环境变量，在“Classpath Entries”部分找到spark-streaming-kafka-0-10_2.11.jar，并且源为“System Classpath”，也作为“Added By User”（似乎是Zeppelin解释器部分的构件）。因此，我的前两个尝试似乎应该可以工作。
我下载了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服务器在同一台机器上使用此端口运行，并且还有一个主题“test”：
%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/analyse/spark/python/lib/pyspark.zip/pyspark/sql/utils.py"， 第63行，在装饰器中： 返回f（* a，** kw）文件“/usr/local/analyse/spark/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py”的行328中，获取返回值： 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（most recent call last）：文件 "/tmp/zeppelin_pyspark-312826888257172599.py"，第380行， exec（code，_zcUserQueryNameSpace）文件“”，第6行，在文件 "/usr/local/analyse/spark/python/lib/pyspark.zip/pyspark/sql/streaming.py"的第400行处加载： return self._df（self._jreader.load（））文件“/usr/local/analyse/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py”的第1257行， 调用 answer，self.gateway_client，self.target_id，self.name）文件“/usr/local/analyse/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/analyse/jar/spark-streaming-kafka-0-10_2.11.jar，但似乎只有在提交程序时才有效。
所以，我也将spark-streaming-kafka-0-10_2.11.jar复制到了/usr/local/analyse/spark/jars/中，那里是spark的所有其他jars。但是，在重新启动（spark和zeppelin）后，我仍然得到相同的错误。
与此同时，我发现我可以在Web浏览器中查看spark的环境变量，在“Classpath Entries”部分找到spark-streaming-kafka-0-10_2.11.jar，并且源为“System Classpath”，也作为“Added By User”（似乎是Zeppelin解释器部分的构件）。因此，我的前两个尝试似乎应该可以工作。