EMR集群PySpark作业直接写S3报“Error closing multipart upload”问题

我之前在处理类似的EMR老版本写入S3的问题时也遇到过这个错误，主要是因为多部分上传配置不合理、S3客户端兼容性或者资源限制导致的。下面是几个经过验证的解决方案，帮你实现直接写入S3：

一、切换到S3A文件系统并调整多部分上传配置 #

EMR 5.x版本推荐使用s3a协议（代替老旧的s3n/s3），它对大文件多部分上传的支持更稳定。同时需要调整以下关键配置：

• 设置S3A实现类：
  spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem
• 增大多部分上传的块大小：默认块太小会导致过多的上传请求，容易超时。建议设置为100MB-1GB（根据集群网络调整）：
  spark.hadoop.fs.s3a.multipart.size=104857600（100MB，单位字节）
• 延长连接和套接字超时时间：避免因网络波动导致的连接中断：
  spark.hadoop.fs.s3a.connection.timeout=300000（5分钟）
  spark.hadoop.fs.s3a.socket.timeout=300000
• 启用快速上传模式：优化多部分上传的并发和重试逻辑：
  spark.hadoop.fs.s3a.fast.upload=true

二、优化DataFrame的输出分区 #

100GB的数据如果分区太少，单个文件会过大，加剧多部分上传的失败概率。建议调整分区数，让每个输出文件控制在1-5GB左右：

• 全局调整Shuffle分区数（如果你的作业有Shuffle操作）：
  spark.sql.shuffle.partitions=100（根据数据量调整，100GB设100-200都可以）
• 直接对DataFrame重分区：

  # 重分区到100个分区，对应每个文件约1GB
  df = df.repartition(100)
  

  注意：repartition会触发全量Shuffle，coalesce不会但只能减少分区数，根据你的数据分布选择。

三、检查IAM权限和网络配置 #

确保EMR集群的EC2实例角色具备以下S3权限，否则可能导致多部分上传无法正常关闭：

• s3:PutObject：写入对象
• s3:AbortMultipartUpload：上传失败时终止多部分任务（非常关键，没有这个权限会直接报"Error closing multipart upload"）
• s3:ListBucket：列出桶内对象

如果集群在VPC中，确保配置了S3网关端点（Gateway Endpoint），避免通过公网访问S3导致的延迟和不稳定。

四、完整PySpark代码示例 #

把上述配置整合到你的作业中：

from pyspark.sql import SparkSession

# 初始化SparkSession并加载S3相关配置
spark = SparkSession.builder \
    .appName("DirectWriteToS3") \
    .config("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") \
    .config("spark.hadoop.fs.s3a.multipart.size", "104857600") \
    .config("spark.hadoop.fs.s3a.connection.timeout", "300000") \
    .config("spark.hadoop.fs.s3a.socket.timeout", "300000") \
    .config("spark.hadoop.fs.s3a.fast.upload", "true") \
    .config("spark.sql.shuffle.partitions", "100") \
    .getOrCreate()

# 假设这里是你的100GB DataFrame加载逻辑
# df = spark.read.parquet("your-input-path")

# 调整分区并写入S3
df.repartition(100).write \
    .mode("overwrite") \
    .parquet("s3a://your-bucket/path/to/output")

spark.stop()

额外排查建议 #

如果还是报错，查看EMR的YARN日志或者Spark driver日志，找到更详细的错误信息：

• 若提示"Connection reset"，说明网络不稳定，进一步增大超时时间或检查VPC配置
• 若提示权限错误，补充对应的IAM权限
• 若单个文件仍然过大，继续增加分区数

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