如何在Spark中指定作业超时时间？（K8s Operator场景）

以下是几种针对执行器丢失导致作业卡住场景的超时配置方案，可根据需求组合使用：

1. 利用Spark内置的网络与心跳超时参数 #

针对执行器丢失的场景，通过调整Spark的网络超时参数，让驱动更快检测到执行器失联并触发清理：

spec:
  sparkConf:
    # 全局网络超时，覆盖所有心跳、RPC通信的超时时间
    spark.network.timeout: "300s"
    # 执行器向驱动发送心跳的间隔，缩短间隔能更快发现执行器丢失
    spark.executor.heartbeatInterval: "30s"
    # 任务失败重试次数，避免因单个执行器丢失导致作业直接失败（可选）
    spark.task.maxFailures: "3"

2. 使用spark-on-k8s-operator的作业超时控制 #

通过operator的activeDeadlineSeconds参数，设置作业的最大运行时长，超时后operator会自动终止驱动和所有执行器Pod：

spec:
  # 作业启动后最长运行5分钟（300秒），超时强制终止
  activeDeadlineSeconds: 300
  # 其他作业配置...

这个参数是Kubernetes原生机制，不管作业是否卡住，只要超过设定时长就会触发清理。

3. 驱动端添加自定义超时逻辑 #

如果需要更灵活的超时控制（比如仅在作业无进展时触发），可以在作业代码中嵌入超时监控逻辑，超时后主动停止SparkContext并终止进程：

import org.apache.spark.SparkContext
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future
import scala.util.control.NonFatal

object TimeControlledSparkJob {
  def main(args: Array[String]): Unit = {
    val sc = SparkContext.getOrCreate()
    val maxRunTimeSeconds = 300

    // 启动超时监控任务
    val timeoutMonitor = Future {
      Thread.sleep(maxRunTimeSeconds * 1000)
      println("作业已达最大运行时长，开始终止所有资源")
      sc.stop()
      System.exit(0)
    }

    try {
      // 这里编写你的Spark作业逻辑
      // ...

      // 作业正常完成后，取消超时监控
      timeoutMonitor.cancel(true)
    } catch {
      case NonFatal(e) =>
        println(s"作业执行异常: ${e.getMessage}")
        timeoutMonitor.cancel(true)
        sc.stop()
        System.exit(1)
    }
  }
}

4. 结合Kubernetes Pod存活探针监控驱动状态 #

给驱动Pod添加存活探针，定期检查驱动进程是否正常运行，多次检测失败后触发终止：

spec:
  driver:
    podSpec:
      containers:
        - name: spark-kubernetes-driver
          livenessProbe:
            exec:
              # 检查驱动进程是否存在
              command: ["pgrep", "-f", "spark.driver.Driver"]
            initialDelaySeconds: 60  # 启动后60秒开始检测
            periodSeconds: 30        # 每30秒检测一次
            failureThreshold: 10     # 连续10次失败则判定驱动异常，终止Pod

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