无法使用火花流从 kafka 主题中读取数据答案

【问题标题】：Unable to read the data from kafka topics using spark streaming无法使用火花流从 kafka 主题中读取数据
【发布时间】：2023-03-18 12:28:01
【问题描述】：

我正在尝试使用火花流从 kafka 主题中读取数据。我能够将消息生成到 kafka 主题中，但是在使用 spark 流从主题中读取数据时，我收到如下错误消息：

ERROR ReceiverTracker: Deregistered receiver for stream 0: Error starting receiver 0 - java.lang.ClassCastException: java.util.HashMap cannot be cast to java.lang.String

下面是代码：

from pyspark import SparkConf, SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
import pprint

conf= SparkConf().setAppName(“streaming test”).setMaster(“yarn-client”)
sc=SparkContext(conf=conf)
ssc=StreamingContext(sc,10)

topic = “newone”
broker = {“metadata.broker.list”: “URL”}
direct = KafkaUtils.createStream(ssc,broker,“test”,{topic:1})

direct.pprint()
ssc.start()
ssc.awaitTermination()

【问题讨论】：

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

【解决方案1】：

DStreams 上的输出操作

print() - 打印每批数据的前十个元素运行流应用程序的驱动程序节点上的 DStream。这是对开发和调试很有用。 Python API 这被称为 Python API 中的pprint()。

java.util.Hashmap 出现在消息中，即 InputDStream[ConsumerRecord[K, V]] 你想打印它，而 java.lang.ClassCastException

你必须解析消息并像this一样打印它...

 direct.transform(...).map(lambda ...)

这里的例子：tests.py

【讨论】：