PyCharm中卡夫卡的电火花流

Question

最近，我一直在尝试调试Pycharm中的pyspark.streaming.kafka类，以便与在linux机器上进行调试相比，更容易排除故障。

下面是我的示例代码：

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils, TopicAndPartition

sc = SparkContext(appName="sample app")
ssc = StreamingContext(sc, 1)
kafkaParams = {"metadata.broker.list": "{broker list}",
               "auto.offset.reset": "smallest"}
kafka_stream = KafkaUtils.createDirectStream(ssc, {topic list}, kafkaParams)

但是，我得到了以下错误：

Traceback (most recent call last):
  File "C:\Program Files (x86)\JetBrains\PyCharm   5.0.3\helpers\pydev\pydevd.py", line 2411, in <module>
    globals = debugger.run(setup['file'], None, None, is_module)
  File "C:\Program Files (x86)\JetBrains\PyCharm    5.0.3\helpers\pydev\pydevd.py", line 1802, in run
    launch(file, globals, locals)  # execute the script
  File "{script path}", line 30, in <module> {topic}], kafkaParams)
  File "C:\spark-1.6.0-bin-  hadoop2.6\python\lib\pyspark.zip\pyspark\streaming\kafka.py", line 152, in  createDirectStream
py4j.protocol.Py4JJavaError: An error occurred while calling o20.loadClass.
: java.lang.ClassNotFoundException:   org.apache.spark.streaming.kafka.KafkaUtilsPythonHelper
    at java.net.URLClassLoader.findClass(Unknown Source)
    at java.lang.ClassLoader.loadClass(Unknown Source)
    at java.lang.ClassLoader.loadClass(Unknown Source)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
    at java.lang.reflect.Method.invoke(Unknown Source)
    at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:381)
    at py4j.Gateway.invoke(Gateway.java:259)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:209)
    at java.lang.Thread.run(Unknown Source)

16/02/22 11:45:49 INFO SparkContext: Invoking stop() from shutdown hook

如果有人能提供一些关于如何在PySpark中调试PyCharm卡夫卡流模块的指导，我将不胜感激。

Answer 1

卡夫卡的支持依赖于外部的spark-streaming-kafka JAR，它不是随火花二进制文件一起提供的。通常，这可以在提交时使用--packages参数指定。

对于使用PyCharm的本地开发，我可以想到的最简单的解决方案是将它添加到$SPARK_HOME/conf/spark-defaults.conf中。假设您使用的是Scala2.10构建的Spark1.6.0：

spark.jars.packages org.apache.spark:spark-streaming-kafka_2.10:1.6.0

请记住，您将无法在Python进程中使用PyCharm调试器。请参阅How can pyspark be called in debug mode?