具有不同架构的 Kafka Spark Streaming json 文件答案

【问题标题】：Kafka Spark Streaming json file with varying schema具有不同架构的 Kafka Spark Streaming json 文件
【发布时间】：2021-08-16 18:26:22
【问题描述】：

我创建了一个管道来使用来自 json 格式的 kafka 主题的流数据。但问题是具有不同模式的 jsons 文件。所以当我在数据帧上应用模式时，它会产生损坏的记录。

如何使用 Pyspark 在 Databricks 中处理 kafka 流中的不同模式

【问题讨论】：

标签： apache-spark apache-kafka databricks spark-structured-streaming

【解决方案1】：

这里有几种方法，具体取决于您的主题中的架构变体类型：

如果所有模式都具有兼容的数据类型（没有具有相同名称但类型不同的列），那么您可以创建一个作为所有模式超集的模式，并在执行from_json 时应用该模式。
如果您的架构不兼容，例如，您的字段名称相同但数据类型不同，那么您可以先将数据读取为文本行，然后根据行类型使用不同的架构进行解码 - 假设所有JSON 具有字符串类型的 type 字段（未测试）：

import pyspark.sql.functions as F
df = spark.readStream.format("kafka")....load().withColumn("value")\
  .withColumn("value", col("value").cast("string"))
df_typed = df.withColumn("jsn", F.from_json(F.col("value"), "type string"))\
  .select("value", "jsn.*")
df1 = df_typed.filter("type == some_value)\
  .withColumn("jsn", F.from_json(F.col("value"), schema1)).select("jsn.*")
# process df1

【讨论】：