我们有Kafka流使用avro。我需要使用python将它连接到sparkstream。
我用了下面的代码：

kvs = KafkaUtils.createDirectStream(ssc, topic, {'bootstrap.servers': brokers}, valueDecoder=decoder)

然后我犯了一个错误。
调用o44.0时出错。
2018-10-11 15:58:01信息日期scheduler:54 - 作业3失败：pythonrdd处的runjob。scala:149，耗时1.403049秒
2018-10-11 15:58:01信息工作scheduler:54 - 已从时间为1539253680000 ms的作业集完成作业流式传输作业1539253680000 ms.0
2018-10-11 15:58:01错误作业scheduler:91 - 运行作业流作业1539253680000 ms时出错。0
org.apache.spark.sparkexception:python:traceback（最近一次调用last）引发了一个异常：
文件“//spark2/python/lib/pyspark.zip/pyspark/streaming/util.py”，第65行，在调用中
r=自身功能（t，*rdds）
文件“//spark2/python/lib/pyspark.zip/pyspark/streaming/dstream.py”，第171行，在takeandprint中
taken=rdd.take（num+1）
文件“//spark2/python/lib/pyspark.zip/pyspark/rdd.py”，第1375行，在take中
res=self.context.runjob（self，takeuptonumleft，p）
文件“//spark2/python/lib/pyspark.zip/pyspark/context.py”，第1013行，在runjob中
sock\u info=self.\u jvm.pythonrdd.runjob（self.\u jsc.sc（），mappedrdd.\u jrdd，分区）
文件“//spark2/python/lib/py4j-0.10.7-src.zip/py4j/java\u gateway.py”，第1257行，在调用中
应答，self.gateway\u client，self.target\u id，self.name）
文件“//spark2/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py”，第328行，在get\u return\u value中
格式（目标id，“.”，名称），值）
py4jjavaerror:调用z:org.apache.spark.api.python.pythonrdd.runjob时出错：org.apache.spark.sparkexception:作业因阶段失败而中止：阶段3.0中的任务0失败了4次，最近一次失败：阶段3.0中的任务0.3丢失（tid 8，gen-cluster\u node，executor 2）：org.apache.spark.sparkexception:无法连接到主题名称1的leader:java.nio.channels.closedchannelexception
然而，在终端上显示这个错误并终止进程之前，我可以使用下面的代码打印rdd

kvs.pprint()

什么是领导？我们怎么过来？