PySpark如何扁平化含动态键的嵌套JSON字符串？

可以实现，核心思路是利用PySpark的Map类型兼容动态键，再提取固定结构的值部分，具体步骤如下：

1. 定义固定值结构的Schema #

因为动态键对应的value结构固定（包含time和amount），先定义这部分的Schema：

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

value_schema = StructType([
    StructField("time", StringType(), nullable=True),
    StructField("amount", IntegerType(), nullable=True)
])

2. 将JSON字符串解析为Map类型 #

把存储JSON的字符串列解析成Map<String, Struct>类型，忽略第一层动态键的随机性：

# 替换"your_json_column"为实际存储JSON的列名
df = df.withColumn("parsed_map", F.from_json("your_json_column", F.map_type(StringType(), value_schema)))

3. 提取Map中的所有值 #

用map_values函数跳过动态键，直接提取所有值并转为数组：

df = df.withColumn("values_array", F.map_values("parsed_map"))

4. 转换为结构体并提取字段 #

根据数据实际情况选择处理方式：

• 如果每条JSON只有一个动态键值对，直接取数组第一个元素转为结构体：

  df = df.withColumn("value_struct", F.element_at("values_array", 1))
  

• 如果每条JSON有多个动态键值对，用explode将数组展开为多行：

  df = df.withColumn("value_struct", F.explode("values_array"))
  

最后提取目标字段并清理中间列：

df = df.withColumn("time", F.col("value_struct.time")) \
       .withColumn("amount", F.col("value_struct.amount")) \
       .drop("your_json_column", "parsed_map", "values_array", "value_struct")

示例说明 #

假设原始JSON列内容为：

{"a1b2c3d4-xxxx-xxxx-xxxx-xxxxxxxxx": {"time": "2024-05-20 12:00:00", "amount": 100}}

经过上述处理后，会生成time（值为2024-05-20 12:00:00）和amount（值为100）两列。

内容的提问来源于stack exchange，提问作者Robert Kossendey