如何在Spark中将DataFrame不同分区写入对应的S3路径？

这个需求完全可以在Spark中实现，以下是两种常用的可落地方案：

方案一：foreachPartition自定义写入（灵活性最高） #

你可以直接通过foreachPartition算子获取每个分区的编号，自行实现写入对应S3路径的逻辑，适配你的1号分区对应路径1、2号对应路径2的映射规则，代码示例如下：

import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.TaskContext
import java.io.OutputStreamWriter

val spark = SparkSession.builder().appName("PartitionCustomWrite").getOrCreate()
// 你的源DataFrame，确认固定为5个分区
val sourceDF = spark.read.parquet("你的源数据路径").repartition(5)

sourceDF.rdd.foreachPartition { partitionDataIter =>
  // 获取当前分区编号，Spark分区编号默认从0开始，+1匹配你的1~5编号规则
  val partitionSn = TaskContext.getPartitionId() + 1
  val targetS3Path = s"s3://你的存储桶/自定义路径前缀${partitionSn}/"

  // 初始化Hadoop文件系统客户端
  val hadoopConf = spark.sparkContext.hadoopConfiguration
  val fs = FileSystem.get(hadoopConf)
  val outputPath = new Path(s"${targetS3Path}/part_${partitionSn}.csv")

  // 写入逻辑可自行替换为parquet、json等格式
  val writer = new OutputStreamWriter(fs.create(outputPath, true))
  partitionDataIter.foreach(row => writer.write(row.mkString(",") + "\n"))
  
  writer.close()
  fs.close()
}

• 优缺点：完全自定义路径命名、写入格式，适配性强；但需要自行处理任务重试的幂等性、文件覆盖、写入一致性等问题。

方案二：动态分区写入（Spark原生功能，稳定性更高） #

如果不需要完全自定义路径前缀，优先用Spark内置的动态分区写入能力，无需手动处理文件IO，代码更简洁：

import org.apache.spark.sql.functions.spark_partition_id

val sourceDF = spark.read.parquet("你的源数据路径").repartition(5)
// 新增分区编号列，值和DataFrame物理分区编号对应
val dfWithPartitionFlag = sourceDF.withColumn("partition_sn", spark_partition_id() + 1)

// 写入时指定按partition_sn分区，自动生成对应编号的文件夹
dfWithPartitionFlag.write
  .mode("overwrite")
  .partitionBy("partition_sn")
  .parquet("s3://你的存储桶/根路径/")

写入完成后生成的路径结构为：

• s3://你的存储桶/根路径/partition_sn=1/
• s3://你的存储桶/根路径/partition_sn=2/
• ...以此类推到partition_sn=5

• 补充：如果不需要路径中的partition_sn=前缀，可以写入完成后调用Hadoop FileSystem API批量重命名文件夹即可。

注意事项 #

• 写入S3时建议开启Hadoop S3A Committer机制，避免出现残留临时文件、数据丢失的一致性问题
• 如果后续DataFrame分区数调整，要同步校验分区编号和S3路径的映射规则，避免写入路径错误

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