Spark作业如何跳过失败步骤 继续处理剩余文件

• 测试作业逻辑：从HDFS路径/tmp/input/下读取1.txt至10.txt共10个文件，生成DataFrame后写入/tmp/output/路径
• 异常模拟设计：处理4.txt时故意写入错误输出路径，预期仅该步骤执行失败，其余文件正常处理
• 实际运行现象：4.txt写入失败后，Spark直接退出上下文，剩余所有文件都无法继续处理
• 核心疑问：Spark是否支持单步骤失败的优雅容错，即记录4.txt失败日志后自动跳过该步骤，继续完成其余文件的读写流程
• 原有实现代码：

inputFileList.foreach(i => {
  try {
    val df = spark.read.csv(s"${i}")
    // Writing to a wrong location should fail the spark job.
    if(count == 4) {
      df.write.format("csv").save(s"$wrongOutput/$count")
    }  else {
      df.write.format("csv").save(s"$output/$count")
    }
    count = count + 1
    println(s"Done: ${count}/${total}")
    println(s"Job ended: ${java.time.LocalDateTime.now()}")
  }

现有写法无法实现容错的核心原因是Spark的执行机制：

• Driver端循环中的try-catch只能捕获Driver侧抛出的异常，而df.write.save()是action算子，实际读写逻辑在Executor端执行
• 默认配置下，子作业执行过程中如果出现错误触发作业失败，会直接导致SparkContext终止，整个应用退出，外层的try-catch无法阻断上下文关停的流程
• Spark本身没有提供全局开关实现自动跳过失败子步骤的容错能力，需要手动做任务隔离实现该效果。

核心思路是将每个文件的读写拆分为独立的子作业，对单个子作业的异常做捕获兜底，避免单个子作业失败拖垮整个SparkContext，具体配置和实现如下：

1. 调整Spark基础配置，降低单任务失败对整个应用的影响：
   • 设置spark.task.maxFailures为合理值，避免单个task失败直接判定整个作业失败
   • 不要配置作业失败自动退出的关联规则，非必要不在子任务逻辑中主动调用spark.stop()
2. 补全异常捕获逻辑，在catch块中记录失败日志即可，不要向上抛出异常
3. 移除循环中共享的可变count变量，直接遍历带索引的文件列表，避免多线程下的变量安全问题

参考实现代码：

import org.apache.spark.sql.SparkSession
import java.time.LocalDateTime

val spark = SparkSession.builder()
  .appName("FaultTolerantFileProcess")
  .config("spark.task.maxFailures", "4")
  .getOrCreate()

val inputBasePath = "/tmp/input/"
val normalOutputPath = "/tmp/output/"
// 模拟错误输出路径
val errorOutputPath = "/tmp/nonexist/wrong/path/"
val fileList = (1 to 10).map(i => s"$inputBasePath$i.txt")

fileList.zipWithIndex.foreach { case (filePath, idx) =>
  val fileNum = idx + 1
  try {
    val df = spark.read.csv(filePath)
    val targetWritePath = if (fileNum == 4) s"$errorOutputPath$fileNum" else s"$normalOutputPath$fileNum"
    df.write.csv(targetWritePath)
    println(s"Process success: $fileNum/${fileList.size}, time: ${LocalDateTime.now()}")
  } catch {
    case e: Exception =>
      println(s"Process failed: $filePath, error: ${e.getMessage}, time: ${LocalDateTime.now()}")
  }
}

spark.stop()

注意：如果失败原因是Executor OOM、集群资源耗尽这类致命级错误，即使做了异常隔离也可能导致SparkContext退出，这类问题需要先做资源调优从根源解决。如果是路径权限不足、路径不存在这类单文件独有的业务错误，上述写法完全可以实现失败自动跳过、剩余任务正常执行的效果。

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