我正在运行这段代码。streaminginputdf是一个关于Kafka主题的readstream。我正在运行spark 2.4.0和kafka 0.10.2
val activityQuery = streamingInputDF.writeStream
.format("console")
.outputMode("append")
.start()
.awaitTermination()我看到这个异常是kafka代理在streaminginputdf上调用writestream时抛出的
[2018-11-24 21:46:42,842] ERROR Closing socket for /127.0.0.1 because of error (kafka.network.Processor)
kafka.common.KafkaException: Wrong request type 18
at kafka.api.RequestKeys$.deserializerForKey(RequestKeys.scala:64)
at kafka.network.RequestChannel$Request.<init>(RequestChannel.scala:50)
at kafka.network.Processor.read(SocketServer.scala:450)
at kafka.network.Processor.run(SocketServer.scala:340)
at java.lang.Thread.run(Thread.java:745)
1条答案
按热度按时间vc6uscn91#
这个问题解决了。我现在能够看到Spark壳中的tweet，因为它们流入Kafka的主题。我所做的只是重启Kafka经纪人，其他一切都没有改变。