Spark嵌套DataFrame扁平化：嵌套字段类型转换后未达预期结构

问题核心 #

处理嵌套的events结构时，由于amount字段同时存在int和double子字段，常规扁平化会拆分出两列，需要合并为单一double类型字段，同时将多子事件（如open）展开为行级别数据。

之前方案失效原因 #

1. match_catalog不生效：读取后的DynamicFrame中events是struct类型，但Catalog定义为map<string, struct<...>>，结构不匹配，无法直接对齐Catalog schema。
2. 路径语法错误：events[].amount是数组类型的路径写法，而实际events是struct，路径无法匹配目标字段。

方案一：Spark DataFrame高效处理（推荐） #

利用Spark内置函数完成结构转换与类型统一，性能更优，适合大数据量场景：

from awsglue.context import GlueContext
from pyspark.context import SparkContext
from pyspark.sql.functions import explode, map_from_entries, array, struct, lit, col, coalesce

sc = SparkContext()
glueContext = GlueContext(sc)

# 从Catalog读取数据
dynamic_frame = glueContext.create_dynamic_frame_from_catalog(
    database="your_database",
    table_name="your_table"
)

# 转为Spark DataFrame
df = dynamic_frame.toDF()

# 1. 将events struct转为map（适配Catalog定义的map类型）
event_fields = df.select("events.*").columns
df = df.withColumn("events_map", map_from_entries(
    array(*[struct(lit(k).alias("key"), col(f"events.{k}").alias("value")) for k in event_fields])
))

# 2. 展开map的value，统一amount字段类型
df_flattened = df.select(
    "id", "hour", "year", "month", "day",
    explode(col("events_map")).alias("event_entry")
).select(
    "id", "hour", "year", "month", "day",
    col("event_entry.value.eventId").alias("eventId"),
    col("event_entry.value.type").alias("type"),
    # 合并int/double为double类型，取第一个非空值
    coalesce(
        col("event_entry.value.amount.double").cast("double"),
        col("event_entry.value.amount.int").cast("double")
    ).alias("amount")
)

# 可选：转回DynamicFrame用于后续Glue操作
final_dynamic_frame = glueContext.create_dynamic_frame.from_df(df_flattened, glueContext, "final_flattened")

方案二：Glue DynamicFrame原生API处理 #

适合纯Glue工作流，无需切换到Spark API：

from awsglue.transforms import Map, Relationalize
from awsglue.context import GlueContext
from pyspark.context import SparkContext

sc = SparkContext()
glueContext = GlueContext(sc)

dynamic_frame = glueContext.create_dynamic_frame_from_catalog(
    database="your_database",
    table_name="your_table"
)

# 自定义映射函数：将events struct转为事件数组，统一amount类型
def transform_events(rec):
    event_list = []
    for event_data in rec["events"].values():
        # 合并int/double为double
        amount_val = None
        if event_data.get("amount", {}).get("double") is not None:
            amount_val = float(event_data["amount"]["double"])
        elif event_data.get("amount", {}).get("int") is not None:
            amount_val = float(event_data["amount"]["int"])
        
        event_list.append({
            "eventId": event_data["eventId"],
            "type": event_data["type"],
            "amount": amount_val
        })
    rec["events"] = event_list
    return rec

# 应用映射
mapped_frame = Map.apply(frame=dynamic_frame, f=transform_events)

# 展开数组并扁平化结构
flattened_frame = Relationalize.apply(
    frame=mapped_frame,
    staging_path="s3://your-staging-bucket/path/",
    name="root"
).select("root")

# 提取目标字段
final_frame = flattened_frame.select_fields(
    ["id", "hour", "year", "month", "day", "eventId", "type", "amount"]
)

方案对比 #

注意事项 #

• 如果原始JSON中events本身是map类型，可直接用explode(map_values(events))展开，无需struct转map步骤。
• 确保Glue Catalog中表的输入格式为json，压缩格式设置为gzip，Glue会自动处理解压。

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