PySpark如何创建DataFrame并提取炸开嵌套元素为多行

核心场景 #

在PySpark on Hive环境中，需从struct类型列sports_info的嵌套字符串数组中提取所有sport_id，最终输出每个business_id对应单个sport_id的多行结构。

步骤1：正确定义Schema并创建测试DataFrame #

你之前的Schema定义缺失外层business_id字段，且ArrayType未指定元素类型，可按如下方式修正：

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

# Hive环境下若已有初始化完成的spark对象可跳过此步
spark = SparkSession.builder.enableHiveSupport().getOrCreate()

# 定义和数据匹配的完整Schema
schema = StructType([
    StructField("business_id", StringType(), True),
    StructField("sports_info", StructType([
        StructField("sports_type", ArrayType(StringType()), True)
    ]), True)
])

# 构造和示例一致的测试数据
test_data = [
    ("abc-123", 
     {"sports_type": [
         # 注意：若原始字符串为键不带双引号的非标准格式，后续解析时需做格式兼容
         '{"sport_name":"most_recent","sport_events":[{"sport_id":568, "val":10.827},{"id":171,"score":8.61}]}'
     ]}
    )
]

df = spark.createDataFrame(data=test_data, schema=schema)
# 打印Schema验证结构
df.printSchema()

执行后输出的正确结构如下：

root
 |-- business_id: string (nullable = true)
 |-- sports_info: struct (nullable = true)
 |    |-- sports_type: array (nullable = true)
 |    |    |-- element: string (containsNull = true)

步骤2：修复UDF逻辑提取所有sport_id #

你之前写的UDF存在变量名未定义、返回类型声明错误、用eval存在安全风险三个问题，修复后的代码如下：

import ast

def extract_sport_ids(sports_info_col):
    res_ids = []
    # 遍历sports_type数组下所有元素（兼容数组多元素场景）
    for sport_str in sports_info_col.sports_type:
        if not sport_str:
            continue
        # 用ast.literal_eval替代eval，避免恶意代码执行风险
        # 若原始字符串为非标准无引号键格式，可在此处加正则替换补全双引号转为合法字典格式
        sport_dict = ast.literal_eval(sport_str)
        event_list = sport_dict.get("sport_events", [])
        for event in event_list:
            # 兼容sport_id、id两种键名，匹配示例中的两个ID值
            sid = event.get("sport_id", event.get("id"))
            if sid is not None:
                res_ids.append(str(sid))
    return res_ids

# 注册UDF，返回值为字符串一维数组
get_sport_ids_udf = F.udf(extract_sport_ids, ArrayType(StringType()))

步骤3：用explode拆分数组为多行得到最终结果 #

调用explode函数把UDF返回的ID数组拆分为单独行，选择需要的字段输出即可：

result_df = df.withColumn("sport_id", F.explode(get_sport_ids_udf(F.col("sports_info")))) \
              .select("business_id", "sport_id")

result_df.show(truncate=False)

执行后输出结果和预期完全一致：

+-----------+--------+
|business_id|sport_id|
+-----------+--------+
|abc-123    |568     |
|abc-123    |171     |
+-----------+--------+

优化提示 #

• 如果sports_type中存储的是标准JSON字符串，优先使用Spark内置的from_json、get_json_object函数做解析，性能比Python UDF高3~10倍，适合TB级大规模数据场景
• 如果上游数据格式可控，建议避免把结构化数据序列化成字符串再存到struct数组里，直接用嵌套Struct/Array类型存储，可以省去字符串解析的开销

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