Pyspark 结构化流 - 来自 2 个嵌套 JSON 的联合数据答案

【问题标题】：Pyspark structured streaming - Union data from 2 nested JSONPyspark 结构化流 - 来自 2 个嵌套 JSON 的联合数据
【发布时间】：2022-01-28 16:08:02
【问题描述】：

我有 2 个 kafka 流数据帧。火花模式如下所示：

root
 |-- key: string (nullable = true)
 |-- pmudata1: struct (nullable = true)
 |    |-- pmu_id: byte (nullable = true)
 |    |-- time: timestamp (nullable = true)
 |    |-- stream_id: byte (nullable = true)
 |    |-- stat: string (nullable = true)

和

root
 |-- key: string (nullable = true)
 |-- pmudata2: struct (nullable = true)
 |    |-- pmu_id: byte (nullable = true)
 |    |-- time: timestamp (nullable = true)
 |    |-- stream_id: byte (nullable = true)
 |    |-- stat: string (nullable = true)

当两个流来自特定批处理窗口时，如何合并所有行？两个流中列的位置相同。每个流都有不同的 pmu_id 值，因此我可以根据该值区分记录。

UnionByName 或 union 从单个数据帧生成流。

我想我需要分解列名，比如this，但这是针对 scala 的。有没有办法在列中自动分解整个 JSON 并将它们合并？

【问题讨论】：

explode 用于数组类型。你正在寻找from_json_object
列pmudata2 是struct 类型，只需使用df.selectExpr("key", "pmudata2.*") 即可得到扁平结构
@blackbishop 我会尝试并告诉你结果..
关于连接，我假设您已经看过文档的这一部分？ spark.apache.org/docs/latest/…
@blackbishop 完美运行。请把它放在答案中。

标签： apache-spark pyspark apache-kafka apache-spark-sql spark-structured-streaming

【解决方案1】：

您只能将explode 函数用于数组和映射类型。在您的情况下，pmudata2 列的类型为 StructType，因此只需使用星号 * 来选择所有子字段，如下所示：

df1 = df.selectExpr("key", "pmudata2.*")

#root
#|-- key: string (nullable = true)
#|-- pmu_id: byte (nullable = true)
#|-- time: timestamp (nullable = true)
#|-- stream_id: byte (nullable = true)
#|-- stat: string (nullable = true)

【讨论】：