读取没有 schema 的 Json Kafka 消息

Question

目前我们正在开发包含 Json 数据的实时数据源。

在阅读以下示例时 - https://sparkbyexamples.com/spark/spark-streaming-with-kafka/< /a>

看起来我们需要一个 kafka json 消息的模式。

有没有其他方法可以在没有模式的情况下处理数据？

Answer 1

运行zookeeper、Kafka服务器和其他所需的服务后尝试下面的代码。

df = spark \
        .readStream \
        .format("kafka") \
        .option("kafka.bootstrap.servers", kafka_bootstrap_servers) \
        .option("subscribe", kafka_topic_name) \
        .option("startingOffsets", "latest")\
        .load()  #earliest

print("Printing Schema of transaction_detail_df: ")
df.printSchema()

transaction_detail_df1 = df.selectExpr("CAST(value AS STRING)")

trans_detail_write_stream = transaction_detail_df1 \
    .writeStream \
    .trigger(processingTime='2 seconds') \
    .option("truncate", "false") \
    .format("console") \
    .start()

trans_detail_write_stream.awaitTermination()

只需更改基本配置，您就可以看到输出

Answer 2

您可以使用 SparkSQL get_json_object 函数从 JSON 字符串数据中解析数据，而无需定义任何其他架构。

您可以简单地使用 cast 函数来反序列化二进制键/值，如示例所示