Spark写入时如何为每个分区生成_SUCCESS文件？

Spark默认仅在写入的根目录生成_SUCCESS文件，要为每个分区目录单独生成该文件，推荐以下两种轻量方案，无需遍历每行数据：

方案一：写入完成后主动创建分区级_SUCCESS文件 #

这种方式先完成常规的分区写入，再通过Hadoop文件系统API为每个涉及的分区创建空的_SUCCESS文件，性能开销极低。

Scala 代码示例 #

import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.sql.functions.col

// 1. 执行原始分区写入操作
df.write.partitionBy("year", "month", "day")
  .mode("append")
  .parquet(table_url)

// 2. 获取本次写入涉及的所有唯一分区
val uniquePartitions = df.select(col("year"), col("month"), col("day"))
  .distinct()
  .collect()

// 3. 获取分布式文件系统实例（适配HDFS/S3等）
val fs = FileSystem.get(spark.sparkContext.hadoopConfiguration)

// 4. 遍历分区，创建_SUCCESS文件
uniquePartitions.foreach { row =>
  val year = row.getAs[Int]("year")
  val month = row.getAs[Int]("month")
  val day = row.getAs[Int]("day")
  val partitionPath = new Path(s"$table_url/year=$year/month=$month/day=$day/_SUCCESS")
  
  // 避免重复创建
  if (!fs.exists(partitionPath)) {
    fs.createNewFile(partitionPath)
    // 可选：同步Spark默认的文件权限
    // fs.setPermission(partitionPath, new org.apache.hadoop.fs.FsPermission("755"))
  }
}

Python 代码示例 #

from pyspark.sql.functions import col
from py4j.java_gateway import java_import

# 1. 执行原始分区写入
df.write.partitionBy("year", "month", "day")
  .mode("append")
  .parquet(table_url)

# 2. 获取本次写入的唯一分区
unique_partitions = df.select(col("year"), col("month"), col("day"))
  .distinct()
  .collect()

# 3. 初始化Hadoop文件系统客户端
java_import(spark._jvm, 'org.apache.hadoop.fs.Path')
java_import(spark._jvm, 'org.apache.hadoop.fs.FileSystem')
fs = spark._jvm.FileSystem.get(spark._jsc.hadoopConfiguration())

# 4. 为每个分区创建_SUCCESS文件
for row in unique_partitions:
    year = row.year
    month = row.month
    day = row.day
    partition_path = spark._jvm.Path(f"{table_url}/year={year}/month={month}/day={day}/_SUCCESS")
    
    if not fs.exists(partition_path):
        fs.createNewFile(partition_path)

方案二：自定义OutputFormat（进阶） #

如果需要长期复用该逻辑，可以自定义Spark的OutputFormat，在每个分区任务完成后自动创建_SUCCESS文件。但该方案需要深入Spark的写入逻辑，实现复杂度较高，适合有定制化需求的场景。

注意事项 #

• 确保Spark的Hadoop配置正确，具备目标文件系统的写权限；
• 若多次追加同一分区，判断文件是否存在的逻辑可以避免重复创建_SUCCESS；
• 该方式仅处理分区级别路径，无需遍历数据行，性能影响可忽略。

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