在AWS Glue 4.0中高效合并多个PySpark DataFrame的最优方案

核心问题分析 #

循环调用union会让Spark的执行计划不断膨胀，每一次union都会在原有计划上叠加新分支，导致后续任务调度和执行效率急剧下降——哪怕单个DataFrame规模很小，累积的开销也会快速失控。

优化方案1：一次性合并所有DataFrame（最直接改进） #

Glue 4.0基于Spark 3.3，支持直接传入DataFrame列表进行批量union，避免循环叠加执行计划。

替换原循环union代码：

from functools import reduce
from pyspark.sql import DataFrame

# 生成所有customer的DataFrame列表后，批量合并
if customer_dfs:
    combined_df = reduce(DataFrame.union, customer_dfs)
else:
    # 处理空列表场景
    combined_df = spark.createDataFrame([], your_default_schema)

或使用更简洁的Spark SQL union函数：

if customer_dfs:
    combined_df = spark.sql.union(customer_dfs)
else:
    combined_df = spark.createDataFrame([], customer_dfs[0].schema if customer_dfs else your_default_schema)

这种方式会生成扁平化的执行计划，性能比循环union提升显著。

优化方案2：避免生成多个小DataFrame（最优解） #

如果业务逻辑允许，不要循环每个customer单独生成DataFrame，而是将所有数据作为整体，通过分组聚合完成每个customer的特定逻辑：

1. 一次性加载所有涉及的数据源，得到包含全量customer数据的大DataFrame
2. 使用groupBy("customer_id")结合agg()、窗口函数（Window.partitionBy("customer_id")）或自定义聚合函数，完成每个customer的转换/聚合
3. 直接得到合并后的结果，无需后续union操作

示例代码框架：

# 一次性加载全量数据
all_data_df = glueContext.create_dynamic_frame.from_catalog(
    database="your_db",
    table_name="your_table"
).toDF()

# 按customer分组执行自定义逻辑
from pyspark.sql import Window
import pyspark.sql.functions as F

window_spec = Window.partitionBy("customer_id")

combined_df = all_data_df.withColumn(
    # 示例：计算每个customer的累计金额
    "customer_total",
    F.sum("amount").over(window_spec)
).withColumn(
    # 示例：计算每个customer的平均值
    "customer_avg",
    F.avg("value").over(window_spec)
)

这种方式完全消除了多个小DataFrame的生成与合并开销，是性能最优的方案，前提是业务逻辑可通过分组/窗口函数实现。

优化方案3：使用Glue DynamicFrame的concat方法 #

Glue的DynamicFrame提供concat方法，可高效合并多个DynamicFrame，内部会做Glue特定优化，适配Glue作业调度体系：

from awsglue.dynamicframe import DynamicFrame

# 将每个customer的DataFrame转换为DynamicFrame
customer_dfs_dynamic = [
    DynamicFrame.fromDF(df, glueContext, f"customer_{i}") 
    for i, df in enumerate(customer_dfs)
]

# 批量合并
combined_dynamic = DynamicFrame.concat(glueContext, customer_dfs_dynamic)
combined_df = combined_dynamic.toDF()

额外性能建议 #

• 确保所有参与合并的DataFrame schema完全一致，避免union时的schema校验开销；若schema可能不一致，用unionByName替代union
• 生成单个customer的DataFrame时，提前过滤无关数据，减少不必要的计算
• 调整Glue作业资源配置（如DPU数量、executor内存），确保有足够资源支撑合并操作

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