如何在S3存储桶的最低层级分区中写入_SUCCESS文件？

我是PySpark/Spark的新手，若问题较为基础还请见谅。我已查阅过StackOverflow上的相关帖子，但未找到可行方案。

当使用Spark提交参数--conf spark.hadoop.mapreduce.fileoutputcommitter.marksuccessfuljobs=true（或代码中设置对应标志）时，Spark会在S3输出写入完成后生成_SUCCESS文件，但该文件仅写入基础路径层级。例如，若写入路径为s3://my-bucket/output_data/region_id=1/marketplace_id=1/（region_id和marketplace_id为分区字段），Spark只会将_SUCCESS文件写入s3://my-bucket/output_data/。

但我负责的应用需要将_SUCCESS文件写入每个marketplace_id分区层级（最低层级分区），以便下游任务确认marketplace_id=1的写入任务已完成，进而启动对应分区的处理任务。

我的初步实现思路如下：

# 创建空DataFrame用于生成_SUCCESS文件
empty_df = spark.createDataFrame([], StructType([]))

# 数据处理与转换
final_df = df1.join(df2, 'inner')

# 将处理后的数据写入指定分区
s3_output_path = 's3://my-bucket/output_data/'
final_df.write.partitionBy('region_id', 'marketplace_id').mode('overwrite').parquet(s3_output_path)

# 现在需要在每个最低层级分区（如marketplace_id=X）写入_SUCCESS文件
# 但我不知道如何获取final_df写入的完整S3路径列表（包含分区路径），无法传入save方法
empty_df.write.format('text').mode('overwrite').save(????)

我需要以下帮助：
Q1：如何获取Spark写入final_df的完整S3路径列表（包含分区路径）？是否必须查询final_df获取region_id和marketplace_id的唯一值？有没有更简便的方法？
Q2：当前的实现思路是否正确？有没有更简便的实现方式？

提前感谢您的帮助与解答！

Q1：获取分区路径的方法 #

最直接可靠的方式是从final_df中提取分区字段的唯一组合——因为Spark写入的分区目录完全由partitionBy指定的字段值生成，无需额外遍历文件系统。具体实现：

# 提取所有唯一的分区字段组合
partition_list = final_df.select('region_id', 'marketplace_id').distinct().collect()

# 构建每个分区的完整S3路径
partition_paths = [
    f"{s3_output_path}region_id={row.region_id}/marketplace_id={row.marketplace_id}"
    for row in partition_list
]

这种方式避免了文件系统操作的延迟、权限问题，直接从数据本身获取分区信息，简洁且可靠。如果一定要通过文件系统获取路径，可借助Spark的Hadoop配置遍历S3目录，但复杂度更高，不推荐。

Q2：实现思路优化与替代方案 #

你的思路方向是对的，但可以做以下优化：

1. 直接创建_SUCCESS文件，避免多余输出
   写入空DataFrame会生成额外的part-*文件，推荐直接通过Hadoop文件系统API创建空的_SUCCESS文件，更高效：

from py4j.java_gateway import java_import
java_import(spark._jvm, 'org.apache.hadoop.fs.Path')

# 获取Hadoop文件系统实例
fs = spark._jvm.org.apache.hadoop.fs.FileSystem.get(spark._jsc.hadoopConfiguration())
success_file = "_SUCCESS"

# 遍历每个分区路径创建_SUCCESS文件
for path_str in partition_paths:
    success_path = spark._jvm.Path(f"{path_str}/{success_file}")
    if fs.exists(success_path):
        fs.delete(success_path, False)
    fs.createNewFile(success_path)

2. 长期复用方案：自定义OutputCommitter
   如果需要大规模复用该功能，可以自定义OutputCommitter，让Spark在每个分区任务完成后自动写入_SUCCESS文件。但这种方式需要理解Spark的提交机制，实现复杂度较高，适合通用化场景。

3. 注意事项

   • 使用overwrite模式时，要确保分区目录确实存在（若某个分区无数据，Spark不会创建对应目录），避免为空目录生成_SUCCESS文件；
   • 操作S3时需确保配置了正确的文件系统实现（如spark.hadoop.fs.s3a.impl），保证文件操作稳定性。

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