Kafka与Spark 2.1结构化流 - 无法反序列化

Question

我想使用 pyspark 将 Kafka (0.10.0.2.5) 作为结构化流的源，版本为 Apache Spark 2.1。

在 Kafka 主题中，我有 JSON 消息（使用 Streamsets Data Collector 推送）。 但是，我无法使用以下代码读取它：

kafka=spark.readStream.format("kafka") \
.option("kafka.bootstrap.servers","localhost:6667") \
.option("subscribe","mytopic").load()
msg=kafka.selectExpr("CAST(value AS STRING)")
disp=msg.writeStream.outputMode("append").format("console").start()

它会生成这个错误：

 java.lang.NoClassDefFoundError: org/apache/kafka/common/serialization/ByteArrayDeserializer

我尝试在readStream行添加：

.option("value.serializer","org.common.serialization.StringSerializer")
.option("key.serializer","org.common.serialization.StringSerializer")

但这并没有解决问题。 有什么想法吗？谢谢您的帮助。

- JS G.

---

尝试使用 org.apache.kafka.common.serialization.StringDeserializer 作为键和值反序列化器。 - Akash Sethi

从这里获取帮助，希望这是你所寻找的内容：https://github.com/akashsethi24/Spark-Kafka-Stream-Example/blob/master/src/main/scala/KafkaConsumer.scala - Akash Sethi

它尝试了，但没有帮助。也许是因为我正在使用结构化流处理。 - JS G.

Answer 1

实际上，我找到了解决方案：我在依赖项中添加了以下jar文件：

spark-streaming-kafka-0-10-assembly_2.10-2.1.0.jar

(从https://mvnrepository.com/artifact/org.apache.spark/spark-streaming-kafka-0-10-assembly_2.10/2.1.0下载后)

- JS G.

---

它完美地工作了！pyspark --jars spark-sql-kafka-0-10_2.11-2.3.2.jar,spark-streaming-kafka-0-10-assembly_2.11-2.3.2.jar让我能够启动我的测试。 - ozw1z5rd