apache-spark - Spark 流 : Reading data from kafka that has multiple schema

Question

我正在为 Spark 流的实现而苦苦挣扎。

来自 kafka 的消息看起来像这样，但有更多的字段

{"event":"sensordata", "source":"sensors", "payload": {"actual data as a json}}
{"event":"databasedata", "mysql":"sensors", "payload": {"actual data as a json}}
{"event":"eventApi", "source":"event1", "payload": {"actual data as a json}}
{"event":"eventapi", "source":"event2", "payload": {"actual data as a json}}

我正在尝试从 Kafka 主题(具有多个模式)读取消息。我需要阅读每条消息并查找事件和源字段并决定将其存储为数据集的位置。实际数据在字段有效负载中作为 JSON，它只是一个记录。

有人可以帮助我实现这个或任何其他替代方案吗？

在同一主题中发送具有多个模式的消息并使用它是一种好方法吗？

提前致谢，

Answer 1

您可以创建一个 Dataframe来自传入的 JSON 对象。

创建 Seq[Sring] JSON 对象。

使用 val df=spark.read.json[Seq[String]] .

对 dataframe df 执行操作你的选择。