gpt4 book ai didi

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

转载 作者:行者123 更新时间:2023-12-04 18:35:27 25 4
gpt4 key购买 nike

我正在尝试利用直接 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)

最佳答案

错误:

java.nio.channels.ClosedChannelException
表示 topic不存在,或无法访问代理或存在某种网络(代理)问题。
通过运行 kafka-console-consumer 确保没有此类连接问题。在 spark 主节点和工作节点上。

关于apache-spark - 使用 PySpark 直接 Kafka Stream (Apache Spark 1.6),我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/35676023/

25 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com