使用 PySpark 直接 Kafka 流 (Apache Spark 1.6)答案

【问题标题】：Direct Kafka Stream with PySpark (Apache Spark 1.6)使用 PySpark 直接 Kafka 流 (Apache Spark 1.6)
【发布时间】：2016-06-11 02:28:45
【问题描述】：

我正在尝试利用直接 kafka 消费者（python 中提供的新功能）从我在 localhost:9092 上运行的自定义 Kafka Producer 捕获数据。

我目前正在使用 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

很遗憾，我遇到了一个无法调试的奇怪错误。任何提示/建议将不胜感激。

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$1.apply(KafkaCluster.scala:366)
        at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$checkErrors$1.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)

【问题讨论】：

标签： apache-spark apache-kafka pyspark spark-streaming

【解决方案1】：

错误：

java.nio.channels.ClosedChannelException

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

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

【讨论】：

我遇到了与@cynical biscuit 相同的问题，并且我确定该主题存在并且主从节点可以访问该主题。但是，我仍然遇到同样的错误。有什么建议吗？
@Decula：如果您的驱动程序和从属服务器中未配置代理的主机名，它也会出现。当您联系 Zookeeper 以获取要读取的偏移范围时，它实际上使用了 kafka 代理的主机名（我在 EMR 上遇到过这个问题）。这应该打印在某处的日志中。寻找那些代理名称并检查驱动程序/从机是否可以解决它！如果没有，在 /etc/hosts 文件中添加主机名-IP 映射
这对我有用。我的问题是我使用了错误的端口。

【解决方案2】：

在我的例子中，火花流来自 kafka 的哪个消费者主题：

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

【讨论】：

【解决方案3】：

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

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

我用了以下

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

【讨论】：