Pyspark处理Event Hub AVRO数据：如何提取嵌套列指定值并展平

方案1：StructType嵌套结构直接提取（性能最优） #

如果目标列是StructType类型的嵌套结构，无需遍历单元格，直接用PySpark内置的字段访问语法即可批量提取指定字段，完全适配分布式执行逻辑：

from pyspark.sql import functions as F

# 示例：嵌套列名为event_body，需提取user_id、event_time、action_type三个字段
flatten_df = df.select(
    F.col("event_body.user_id").alias("user_id"),
    F.col("event_body.event_time").alias("event_time"),
    F.col("event_body.action_type").alias("action_type")
    # 需保留的非嵌套字段也可直接加在此处
)

多层嵌套场景直接写全字段路径即可，比如event_body.user_info.addr.city。

方案2：ArrayType嵌套结构先展开再提取 #

如果嵌套列是数组套结构体的结构，先用explode把数组拆分为单行单元素，再提取指定字段：

# 先拆分数组为多行
exploded_df = df.withColumn("exploded_item", F.explode(F.col("event_body.item_list")))

# 提取数组元素内的指定字段
flatten_df = exploded_df.select(
    F.col("exploded_item.item_id").alias("item_id"),
    F.col("exploded_item.item_name").alias("item_name")
)

方案3：多字段批量提取（减少重复代码） #

如果需要提取的指定字段较多，可以提前定义字段映射列表，批量生成提取逻辑：

# 定义映射规则：(嵌套字段全路径, 输出列名)
extract_fields = [
    ("event_body.user_id", "user_id"),
    ("event_body.event_time", "event_time"),
    ("event_body.action.type", "action_type"),
    ("event_body.device.info", "device_info")
]

# 批量生成选择列，可按需追加原有非嵌套字段
select_cols = [F.col(path).alias(col_name) for path, col_name in extract_fields]
flatten_df = df.select(*select_cols)

• 以上所有方案均为PySpark原生执行逻辑，性能远高于自定义UDF或逐行遍历处理，适配Event Hub导出的大规模AVRO数据场景。
• 遇到空值场景可搭配cast()做类型转换，或用coalesce()指定默认值。

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