如何使AWS Glue作业以错误状态终止或标记为失败？

我最近碰到一个需求：在AWS Glue作业里，当读取的动态帧行数小于等于设定阈值时，得让作业直接以错误状态终止——既不能跳过后续逻辑（那样作业会显示成功，不符合预期），也不想靠抛出异常的方式让作业失败（虽然能达到失败效果，但总觉得不够可控）。想找一种不抛出异常就能让作业错误终止的方法。

先说说我最初踩过的坑，当时想着直接用SparkContext来终止作业：

val spark: SparkContext = SparkContext.getOrCreate()
val glueContext: GlueContext = new GlueContext(spark)
val jobId = GlueArgParser.getResolvedOptions(sysArgs, Seq("JOB_ID").toArray)("JOB_ID")
spark.cancelJob(jobId)

但很快就发现两个致命问题：

• SparkContext属于Glue内部依赖的框架层对象，直接调用它的终止方法可能导致不可预测的不稳定结果，比如资源清理不彻底、残留进程之类的问题；
• org.apache.spark.SparkContext#cancelJob要求传入的是Int类型的作业ID，而AWS Glue的JOB_ID是长字符串格式（比如j_aaa11111a1a11a111a1aaa11a11111aaa11a111a1111111a111a1a1aa111111a），根本没法直接传入，这个方法完全不适用。

那到底怎么正确实现需求呢？其实AWS Glue本身就提供了官方的可控终止方法，用Glue的Job工具类就能搞定：

首先确保你的作业里已经初始化了Job对象（一般Glue作业模板里都会包含这部分），然后在需要终止的逻辑里调用Job.failJob()方法，传入自定义的错误信息即可。完整的代码示例大概是这样的：

import com.amazonaws.services.glue.util.Job
import com.amazonaws.services.glue.GlueContext
import org.apache.spark.SparkContext

object MyGlueJob {
  def main(sysArgs: Array[String]): Unit = {
    val spark: SparkContext = SparkContext.getOrCreate()
    val glueContext: GlueContext = new GlueContext(spark)
    
    // 初始化Job，这一步是基础，必须先做
    val args = GlueArgParser.getResolvedOptions(sysArgs, Seq("JOB_NAME", "JOB_ID").toArray)
    Job.init(args("JOB_NAME"), glueContext, args)
    
    // 读取数据源
    val input = glueContext
      .getCatalogSource(database = "my_db", tableName = "my_table")
      .getDynamicFrame()
    
    val myLimit = 10
    if (input.count() <= myLimit) {
      // 以错误状态终止作业，自定义错误信息方便排查
      Job.failJob(s"Input record count (${input.count()}) is less than or equal to threshold $myLimit, terminating job with error.")
    }
    
    // 后续正常执行的业务逻辑
    // ...
    
    // 正常完成时提交作业状态
    Job.commit()
  }
}

这个方法的核心优势在于：

• 完全符合Glue的作业生命周期管理规范，是官方推荐的方式，不会有不稳定的隐患；
• 不需要抛出异常，直接通过框架API标记作业为失败状态；
• 可以自定义清晰的错误信息，方便在Glue控制台快速定位失败原因。

另外补充一下，如果你不排斥抛异常的方式，主动抛出一个自定义RuntimeException也能让作业失败，但如果想要更可控、更符合Glue作业规范的做法，上面的Job.failJob()就是最优解。

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