使用 PySpark (Apache Spark 1.6) 直接 Kafka Stream

Question

我正在尝试利用直接的 kafka 消费者（python 中可用的新功能），从我运行在 [=32= 上使用的自定义 Kafka 生产者捕获数据].

我目前正在使用 spark 1.6 示例脚本提供的 "direct_kafka_wordcount.py"。

来源: https://github.com/apache/spark/blob/master/examples/src/main/python/streaming/direct_kafka_wordcount.py

文档： http://spark.apache.org/docs/latest/streaming-kafka-integration.html

我在运行程序中使用以下命令：

    ~/spark-1.6.0/bin/spark-submit --jars 
    ~/spark-1.6.0/external/kafka-assembly/target/spark-streaming-kafka-assembly_2.10-1.6.0.jar           
direct_kafka_wordcount.py localhost:9092 twitter.live

不幸的是，我遇到了一个我无法调试的奇怪错误。任何 tips/suggestions 将不胜感激。

py4j.protocol.Py4JJavaError: An error occurred while calling o24.createDirectStreamWithoutMessageHandler.
: org.apache.spark.SparkException: java.nio.channels.ClosedChannelException
        at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$checkErrors.apply(KafkaCluster.scala:366)
        at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$checkErrors.apply(KafkaCluster.scala:366)
        at scala.util.Either.fold(Either.scala:97)
        at org.apache.spark.streaming.kafka.KafkaCluster$.checkErrors(KafkaCluster.scala:365)
        at org.apache.spark.streaming.kafka.KafkaUtils$.getFromOffsets(KafkaUtils.scala:222)
        at org.apache.spark.streaming.kafka.KafkaUtilsPythonHelper.createDirectStream(KafkaUtils.scala:720)
        at org.apache.spark.streaming.kafka.KafkaUtilsPythonHelper.createDirectStreamWithoutMessageHandler(KafkaUtils.scala:688)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:606)
        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(Thread.java:745)

Answer 1

错误：

java.nio.channels.ClosedChannelException

表示 topic 不存在，或者无法访问代理，或者存在某种网络（代理）问题。

通过运行 kafka-console-consumer 确保 spark 主节点和工作节点上没有此类连接问题。

Answer 2

在我的案例中，来自 kafka 的 spark streaming 消费者主题：

出现错误，程序退出。所以我检查 metadata.broker.list，只添加了一个经纪人。添加除其中一个之外的所有经纪人。一切顺利，但仍然得到警告 org.apache.spark.SparkException: java.nio.channels.ClosedChannelException ，所以我从 zk 检查 kafka brokers 状态，发现一个 broker 坏了导致这样的错误。

Answer 3

我有类似的问题。但结果是不同的解决方案。我有不同版本的 scala 运行用于 spark 和 kafka。

我最终在双方都使用了相同的版本，然后 pyspark 能够生成类。

我用了以下

火花：spark-1.6.3-bin-hadoop2.6.tgz 火花流卡夫卡：spark-streaming-kafka-assembly_2.10-1.6.3.jar

使用 PySpark (Apache Spark 1.6) 直接 Kafka Stream

Direct Kafka Stream with PySpark (Apache Spark 1.6)

apache-kafka

apache-spark

spark-streaming

pyspark