PySpark DataFrame如何从存储字典列表的列提取字段生成新列

你的trans列属于两层嵌套结构：外层是数组类型，数组内存储结构体（字典），结构体内的group字段又是一层数组嵌套结构体，直接用Spark内置函数拆解即可，无需自定义UDF，数千条数据处理性能无压力。

核心处理逻辑 #

• 若存在单条记录对应多个trans元素、多个group元素的情况，用explode函数炸开数组，保证所有元素都能被提取：

from pyspark.sql.functions import col, explode

result_df = (
    df
    # 炸开最外层trans数组，将数组内每个字典转为独立行
    .withColumn("trans_item", explode(col("trans")))
    # 炸开嵌套的group数组，提取内部字段
    .withColumn("group_item", explode(col("trans_item.group")))
    .select(
        "id",
        col("group_item.sub").alias("sub"),
        col("group_item.pCode").alias("pCode"),
        col("group_item.dtl").alias("dtl"),
        col("group_item.reas").alias("reas"),
        col("trans_item.description").alias("description"),
        col("trans_item.DType").alias("DType")
    )
)

• 若你确认每条记录的trans数组、group数组都固定只有1个元素，可以直接按索引取值，跳过explode步骤，执行效率更高：

result_df = df.select(
    "id",
    col("trans")[0]["group"][0]["sub"].alias("sub"),
    col("trans")[0]["group"][0]["pCode"].alias("pCode"),
    col("trans")[0]["group"][0]["dtl"].alias("dtl"),
    col("trans")[0]["group"][0]["reas"].alias("reas"),
    col("trans")[0]["description"].alias("description"),
    col("trans")[0]["DType"].alias("DType")
)

补充说明 #

你给出的期望输出中reas列为0.77、0.137这类浮点值，和原始数据中存储的字符串数组格式不一致，这部分是自定义映射逻辑，可以在select阶段通过when匹配规则实现转换，示例代码如下：

from pyspark.sql.functions import when

result_df = result_df.withColumn(
    "reas",
    when(col("reas")[0] == "ZYD", 0.77)
    .when(col("reas")[0] == "XCD", 0.137)
    .when(col("reas")[0] == "KYD", 0.317)
    .when(col("reas")[0] == "VVD", 0.457)
    .when(col("reas")[0] == "BYD", 0.947)
)

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