如何获取当前SparkContext的jobId以实现编程取消Spark任务

Spark JobId是每个Action操作触发的作业对应的唯一标识，和大家通常熟悉的ApplicationId属于不同维度的标识，常见获取方式分以下两类场景：

1. 作业提交端主动获取 #

1.1 单作业同步获取 #

提交作业后通过SparkStatusTracker查询最新生成的JobId即可：

• Scala示例

val sc = spark.sparkContext
// 提交Action前记录当前已有的Job数量
val preJobNum = sc.statusTracker.getJobIdsForGroup(null).size
// 执行你的业务Action操作，比如count、saveAsTextFile等
dataRdd.count()
// 取最新生成的JobId
val allJobIds = sc.statusTracker.getJobIdsForGroup(null)
val currentJobId = if (allJobIds.size > preJobNum) allJobIds.last else -1

• Python示例

sc = spark.sparkContext
pre_job_num = len(sc.statusTracker().getJobIdsForGroup(None))
data_rdd.count()
all_job_ids = sc.statusTracker().getJobIdsForGroup(None)
current_job_id = all_job_ids[-1] if len(all_job_ids) > pre_job_num else -1

1.2 批量作业按组获取 #

如果是批量提交多个关联作业，可以提前给作业设置自定义JobGroup，后续直接通过Group获取所有关联JobId，也可以直接取消整组作业：

val sc = spark.sparkContext
// 第一个参数为自定义的GroupId，第二个为描述信息
sc.setJobGroup("batch_calc_group", "月度数据批量计算")
// 执行多个关联的Action操作
orderRdd.saveAsTable("dwd.dwd_order_df")
userRdd.count()
// 获取该Group下所有JobId
val jobIds = sc.statusTracker.getJobIdsForGroup("batch_calc_group")
// 也可以直接取消整组作业，无需逐个传JobId
sc.cancelJobGroup("batch_calc_group")

提示：仅需取消同组所有作业的场景，优先调用cancelJobGroup方法，不需要单独获取每个JobId再逐一取消。

2. 作业运行时内部获取 #

如果需要在作业运行过程中（比如自定义UDF、算子逻辑内）获取当前所属的JobId，可以通过TaskContext直接读取：

• Scala示例

import org.apache.spark.TaskContext

val processedRdd = dataRdd.mapPartitions(iter => {
  val taskCtx = TaskContext.get()
  val currentJobId = taskCtx.jobId()
  // 可插入自定义逻辑，比如异常阈值触发时调用取消接口
  iter
})

• Python示例

from pyspark import TaskContext

def process_part(iter):
    task_ctx = TaskContext.get()
    current_job_id = task_ctx.jobId()
    # 自定义业务逻辑
    return iter

processed_rdd = data_rdd.mapPartitions(process_part)

拿到JobId后，直接调用取消接口即可终止对应作业：

spark.sparkContext.cancelJob(currentJobId)

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