如何监控Spark Job生成的文件数量并在文件过多时抛出异常

以下是不同场景下的实现方式，可按需选择：

1. 写入完成后实际统计（精度最高） #

适合写入完成后做最终校验的场景，兼容所有存储类型（HDFS、S3、本地文件系统等）：

• 核心逻辑是调用Hadoop FileSystem API遍历输出路径，过滤掉_SUCCESS等非数据文件后统计数量

// Scala 示例代码
import org.apache.hadoop.fs.{FileSystem, Path}

val outputPath = "你的文件输出路径"
val hadoopConf = spark.sparkContext.hadoopConfiguration
val fs = FileSystem.get(hadoopConf)
// 只统计part开头的实际数据文件
val actualFileCount = fs.listStatus(new Path(outputPath))
  .filter(_.isFile)
  .count(_.getPath.getName.startsWith("part-"))

// 超过阈值抛出异常
val maxAllowCount = 1000 // 按你的业务需求调整阈值
if (actualFileCount > maxAllowCount) {
  throw new RuntimeException(s"输出文件数超出阈值，当前生成${actualFileCount}个，最大允许${maxAllowCount}个")
}

2. 写入前预校验（性能最优） #

不用等待写入完成即可提前拦截异常，适合常规无特殊写入规则的场景：

Spark默认生成的输出文件数和写入前最后一个RDD/DataFrame的分区数完全一致，只要你没有开启动态分区、分桶写入等特殊配置，该预估值和实际值完全相等

val df = // 你要写入的DataFrame
val estimateFileCount = df.rdd.getNumPartitions

val maxAllowCount = 1000
if (estimateFileCount > maxAllowCount) {
  throw new RuntimeException(s"预估算输出文件数超出阈值，当前分区数为${estimateFileCount}，最大允许${maxAllowCount}个")
}
// 校验通过再执行写入逻辑
df.write.parquet(outputPath)

3. 动态分区写入场景预校验 #

如果使用Spark SQL动态分区写入，可通过以下公式预估算文件数：输出文件数 = 动态分区去重数量 * DataFrame原始分区数

val partitionCol = "dt" // 你的动态分区字段名
val distinctPartitionNum = df.select(partitionCol).distinct().count()
val rddPartitionNum = df.rdd.getNumPartitions
val estimateFileCount = distinctPartitionNum * rddPartitionNum

val maxAllowCount = 1000
if (estimateFileCount > maxAllowCount) {
  throw new RuntimeException(s"动态分区场景预估算输出文件数超出阈值，预估值为${estimateFileCount}，最大允许${maxAllowCount}个")
}

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