AWS Glue按指定大小分区DynamicFrames失败，求排查与优化方案

一、现有方案问题原因分析 #

方案1报错原因 #

你计算的分区数过大，触发了Spark shuffle操作的资源瓶颈：

• repartition(num)是全量shuffle操作，会将所有数据打乱重分区，当分区数远超集群Executor的承载能力（比如单Executor能处理的分区数有限），会导致任务堆积、内存溢出，最终触发Py4J调用错误。
• 硬编码分区数为2时，shuffle任务量极小，集群能轻松处理，所以正常运行。

方案2问题原因 #

• boundedSize是Parquet读取阶段的拆分参数，用于控制读取大Parquet文件时的拆分块大小，对CSV输出的文件大小没有直接约束作用，所以输出文件大小远超过10MB。
• 仅含表头的空文件是因为Glue读取Parquet时生成了空分区（比如原Parquet文件有空分片、或者Crawler生成的表包含空分区元数据），DynamicFrame输出CSV时，即使分区无数据也会写入表头。

二、生产级适配解决方案 #

针对数百GB级数据，需要兼顾分区合理性、避免全量内存加载、严格控制输出文件大小，推荐以下步骤：

1. 精准计算合理分区数 #

不要仅按总数据量/10MB计算分区数，需结合集群资源（Executor数量、单Executor内存/CPU）调整，公式参考：

# 预估总数据量（可通过Glue表的统计信息或S3存储大小获取）
total_data_size = 总数据字节数
# 单分区目标大小（设为10MB，即10*1024*1024=10485760字节）
target_partition_size = 10*1024*1024
# 基础分区数
base_num_partitions = total_data_size // target_partition_size
# 结合集群Executor总核心数调整（建议不超过核心数的2倍，避免资源浪费）
executor_total_cores =  Worker数量 * 单Worker核心数
num_partitions = min(base_num_partitions, executor_total_cores * 2)
# 至少保留1个分区
num_partitions = max(num_partitions, 1)

2. 使用Spark API高效分区并输出 #

跳过DynamicFrame转DataFrame再转回的冗余操作，直接用Spark DataFrame处理，避免Glue DynamicFrame的额外开销，同时控制输出文件大小：

import sys
from awsglue.context import GlueContext
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session

# 从Glue Catalog加载表为Spark DataFrame
df = spark.table("your_database.your_table")

# 计算合理分区数（替换为实际计算逻辑）
total_data_size = 100 * 1024 * 1024 * 1024  # 示例：100GB
target_size = 10 * 1024 * 1024
executor_total_cores = 20  # 示例：10个G.2X Worker，每个2核心
num_partitions = max(1, min(total_data_size // target_size, executor_total_cores * 2))

# 优先用coalesce（无shuffle）调整分区，仅当需要增加分区时用repartition
if df.rdd.getNumPartitions() > num_partitions:
    df = df.coalesce(num_partitions)
else:
    df = df.repartition(num_partitions)

# 输出CSV，通过记录数控制文件大小
df.write \
  .option("header", "true") \
  .option("maxRecordsPerFile", 100000)  # 根据单条记录平均大小调整，比如100字节/条时，10万条约10MB
  .mode("overwrite") \
  .csv("s3://your-output-bucket/path/")

• maxRecordsPerFile：通过控制每个文件的记录数间接控制文件大小，可根据实际单条记录的平均字节数灵活调整。
• 优先使用coalesce（无shuffle）减少分区，仅在需要增加分区时使用repartition（含shuffle），降低资源消耗。

3. 消除空表头文件 #

可以通过两种方式避免生成仅含表头的空文件：

• 方式1：过滤空数据（针对数据本身存在空分区的情况）

# 替换为表中的非空关键字段，过滤掉无有效数据的行
df = df.filter(df["your_key_column"].isNotNull())

• 方式2：设置Spark输出参数（Spark 2.4+版本支持）

df.write \
  .option("header", "true") \
  .option("maxRecordsPerFile", 100000) \
  .option("ignoreEmptyFiles", "true")  # 忽略空文件，不写入S3
  .mode("overwrite") \
  .csv("s3://your-output-bucket/path/")

4. 生产环境资源优化 #

• 调整Glue Job的Executor配置：根据数据量选择合适的Worker type（比如G.2X/G.4X大内存机型）和Number of workers，避免资源不足或浪费。
• 开启Spark动态资源分配：在Glue Job的Spark configuration中添加以下参数，自动根据任务需求调整资源：

--conf spark.dynamicAllocation.enabled=true
--conf spark.shuffle.service.enabled=true

• 避免全量缓存：不要手动调用cache()或persist()，依赖Spark的按需加载机制（分区裁剪、谓词下推），减少内存占用。

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