如何在PySpark中处理复杂数据类型：将嵌套属性转为独立列

问题背景 #

PySpark DataFrame中有一个字符串类型的attributes列，存储内容为JSON数组，每个元素包含name、type、value三个字段。需要将这些数据转换为以name作为列名、对应value作为列值的结构化DataFrame。

解决方案步骤 #

以下是完整实现步骤，包含静态提取指定列和动态提取所有列两种方式：

1. 导入依赖并定义Schema

先导入PySpark相关函数，同时定义解析attributes列的Schema：

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, map_from_entries, col, explode
from pyspark.sql.types import ArrayType, StructType, StructField, StringType

# 定义attributes列的解析Schema
attr_schema = ArrayType(
    StructType([
        StructField("name", StringType(), nullable=True),
        StructField("type", StringType(), nullable=True),
        StructField("value", StringType(), nullable=True)
    ])
)

2. 解析字符串列为结构化数组

将原始字符串类型的attributes列解析为PySpark可操作的数组结构：

# 假设你的DataFrame名为df
df = df.withColumn("attributes_parsed", from_json(col("attributes"), attr_schema))

3. 将数组转换为键值对Map

使用map_from_entries函数，把数组中的每个元素（包含name和value）转换为Map结构，键为name，值为value：

df = df.withColumn("attr_map", map_from_entries(col("attributes_parsed")))

4. 提取目标列

方式一：静态提取指定列

如果只需要提取部分列，直接从Map中选择对应键并设置别名：

df_result = df.select(
    col("attr_map")["sfcc.created_by"].alias("sfcc.created_by"),
    col("attr_map")["shippingLines"].alias("shippingLines"),
    col("attr_map")["sourceChannel"].alias("sourceChannel"),
    col("attr_map")["leaveAtDoor"].alias("leaveAtDoor")
    # 按需添加其他需要的列
)

方式二：动态提取所有列

如果需要提取所有name对应的列，可以先获取所有唯一的name值，再动态生成选择表达式：

# 获取所有唯一的name字段
all_attr_names = df.select(explode(col("attributes_parsed.name"))).distinct().rdd.flatMap(lambda x: x).collect()

# 动态生成列选择列表
select_cols = [col("attr_map")[name].alias(name) for name in all_attr_names]

# 生成最终结果DataFrame
df_result = df.select(*select_cols)

5. （可选）解析JSON类型的value

对于type为JSON的字段（如shippingLines、splitPayments），可以进一步将其字符串值解析为对应的结构化类型：

from pyspark.sql.types import MapType, ArrayType, StructType, StructField

# 示例：解析shippingLines为Map类型
shipping_lines_schema = MapType(StringType(), StringType())
df_result = df_result.withColumn("shippingLines", from_json(col("shippingLines"), shipping_lines_schema))

# 示例：解析splitPayments为数组结构
split_payments_schema = ArrayType(
    StructType([
        StructField("price", StringType()),
        StructField("gateway", StringType()),
        StructField("paymentGatewayName", StringType())
    ])
)
df_result = df_result.withColumn("splitPayments", from_json(col("splitPayments"), split_payments_schema))

结果示例 #

处理后的DataFrame将呈现如下结构（以部分列为例）：

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