Scala Spark中retry方法不抛异常实现及编译类型不匹配错误解决

你定义的retry函数声明返回值为泛型T，但最后一次重试失败的分支仅执行了println("error")，该语句返回值类型是Unit，和要求的T类型不匹配，因此触发类型不兼容编译错误。
另外原代码还存在隐藏逻辑问题：Try { return fn }里的return会直接跳出整个retry函数，异常不会被后续的match逻辑捕获，重试逻辑完全不会生效。

可以根据你的业务场景选择以下两种修改方式：

方案1：返回Option[T]类型（推荐） #

这种实现可以明确区分执行成功/失败状态，调用侧可直接通过模式匹配处理两种场景，符合Scala函数式编程规范：

import scala.util.{Try, Success, Failure}

def retry[T](n: Int)(fn: => T): Option[T] = {
  Try(fn) match {
    case Success(x) => Some(x)
    case Failure(f) =>
      Thread.sleep(3000)
      if (n > 1) {
        retry(n - 1)(fn)
      } else {
        // 自定义错误处理
        println(s"全部${n}次重试失败，错误信息：${f.getMessage}")
        None
      }
  }
}

调用示例：

// 调用时直接匹配状态即可
retry(3) {
  // 你的业务逻辑
  spark.read.parquet("/path/to/file")
} match {
  case Some(df) => df.show()
  case None => println("读取文件失败，跳过该步骤")
}

方案2：支持传入失败默认值，返回T类型 #

如果业务场景允许重试失败后返回一个约定好的默认值，可以新增默认值入参：

import scala.util.{Try, Success, Failure}

def retry[T](n: Int, fallback: T)(fn: => T): T = {
  Try(fn) match {
    case Success(x) => x
    case Failure(f) =>
      Thread.sleep(3000)
      if (n > 1) {
        retry(n - 1, fallback)(fn)
      } else {
        println(s"全部${n}次重试失败，返回默认值，错误信息：${f.getMessage}")
        fallback
      }
  }
}

• 如果重试逻辑在Spark Executor端执行，不建议使用Thread.sleep，长时间sleep可能导致Executor心跳超时被Driver判定为失效，引发任务额外重试。
• 如果你的业务逻辑本身是Spark的转换算子（无实际执行逻辑），建议将重试逻辑作用在行动算子上，避免无意义的重试。

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