AWS Glue PySpark迁移DynamoDB到S3遇磁盘空间不足及Shuffle配置咨询

一、核心Spark Shuffle配置调整 #

直接在代码里通过SparkConf设置以下参数，控制Shuffle的磁盘占用、内存使用，避免本地磁盘耗尽：

• spark.shuffle.file.buffer：调大Shuffle写缓冲区（默认32k，改成64k/128k），减少磁盘IO次数
• spark.reducer.maxSizeInFlight：缩小Reducer单次拉取的Shuffle数据量（默认48m，改成24m），降低单节点磁盘压力
• spark.shuffle.spill.compress：确保开启Shuffle溢出数据压缩（默认true，显式设置避免意外关闭）
• spark.shuffle.compress：开启Shuffle输出压缩（默认true），减少磁盘存储体积
• spark.sql.shuffle.partitions：减少Shuffle分区数（默认200，6GB数据改成32/64即可），减少临时小文件数量
• spark.dynamicAllocation.enabled：Glue 2.0+版本开启动态资源分配，自动调整Worker数量分散压力

二、优化作业执行逻辑 #

1. 移除Filter函数里的print语句，避免日志占用额外磁盘空间
2. 简化Filter判断逻辑，减少不必要的计算开销

三、修改后的完整代码 #

from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.transforms import Filter
from pyspark import SparkConf

# 配置Spark Shuffle参数，解决磁盘空间不足问题
conf = SparkConf() \
    .set("spark.shuffle.file.buffer", "64k") \
    .set("spark.reducer.maxSizeInFlight", "24m") \
    .set("spark.shuffle.spill.compress", "true") \
    .set("spark.shuffle.compress", "true") \
    .set("spark.sql.shuffle.partitions", "64") \
    .set("spark.dynamicAllocation.enabled", "true")

# 用配置好的SparkConf初始化上下文
sc = SparkContext.getOrCreate(conf=conf)
glue_context = GlueContext(sc)

# 读取DynamoDB数据
resources_table_dynamic_frame = glue_context.create_dynamic_frame.from_options(
    connection_type="dynamodb",
    connection_options={
        "dynamodb.input.tableName": "my_table",
        "dynamodb.throughput.read.percent": "0.4",
        "dynamodb.splits": "8"
    }
)

# 精简后的Filter逻辑
def filter_new_id(dynamicRecord):
    uri = dynamicRecord['Uri']
    internal_id = uri.split(":")[1]
    return internal_id != dynamicRecord['id']

resource_with_old_id = Filter.apply(
    frame=resources_table_dynamic_frame,
    f=lambda x: filter_new_id(x),
    transformation_ctx='resource_with_old_id'
)

# 写入S3，按需设置分区键减少临时文件
glue_context.write_dynamic_frame_from_options(
    frame=resource_with_old_id,
    connection_type="s3",
    connection_options={
        "path": "s3://path/",
        "partitionKeys": []
    },
    format="json"
)

四、额外注意点 #

• 如果用的是Glue 1.0版本，动态资源分配需要额外配置，建议升级到2.0+版本更省心
• 若调整配置后仍有问题，可适当增加Worker数量，把磁盘压力分散到多个节点
• 查看CloudWatch日志里的磁盘使用指标，定位具体是读取、Shuffle还是写入阶段占满了磁盘

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