PySpark DataFrame嵌套列展开：替代BigQuery UNNEST实现

错误原因 #

你之前的PySpark SQL报错，是因为array_element是结构体类型，而explode函数仅支持数组或Map类型作为输入。BigQuery里的UNNEST([element.array_element])是将单个结构体包装成数组后再展开，PySpark中需要用array()函数实现相同的包装逻辑。

PySpark SQL等价写法 #

SELECT 
    event_name, 
    event_params,
    (SELECT 
        array_element.value.int_value
     FROM 
        explode(event_params.bag) AS element,
        explode(array(element.array_element)) AS array_element
     WHERE 
        array_element.key = 'ga_session_id'
        AND array_element.value.int_value IS NOT NULL
     LIMIT 1) AS ga_session_id
FROM tbl_name

更简洁的DataFrame API实现 #

如果倾向于用DataFrame API而非SQL，可使用更直观的嵌套结构处理方式：

from pyspark.sql import functions as F

# 读取原表
df = spark.table("tbl_name")

# 提取ga_session_id
result_df = df.withColumn(
    "ga_session_id",
    F.expr("""
        filter(
            transform(event_params.bag, x -> x.array_element),
            x -> x.key = 'ga_session_id' AND x.value.int_value IS NOT NULL
        )[0].value.int_value
    """)
).select("event_name", "event_params", "ga_session_id")

# 查看结果
result_df.show()

代码说明 #

• transform(event_params.bag, x -> x.array_element)：遍历bag数组，提取每个元素中的array_element结构体，生成新数组
• filter(...)：筛选出key为ga_session_id且int_value非空的元素
• [0].value.int_value：取第一个符合条件的元素的int_value，对应BigQuery中的LIMIT 1逻辑

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