如何根据Spark程序返回的退出码跳过Airflow SparkSubmitOperator下游任务

你的Spark提交任务可以通过以下两种方式实现返回特定退出码时自动跳过下游任务：

方案1：自定义SparkSubmitHook与SparkSubmitOperator #

• 继承原生SparkSubmitHook，重写返回码校验逻辑：原生Hook会对所有非0返回码抛出异常，你可以预设几个触发跳过的退出码，当Spark任务返回对应退出码时，不抛执行失败异常，而是抛出AirflowSkipException，此时当前任务会被标记为跳过，所有下游任务默认自动跳过。
• 若需要更灵活的分支逻辑，可以不直接抛跳过异常，而是将Spark返回的退出码推送至XCom，下游搭配BranchPythonOperator读取返回码判断分支走向。

方案2：改用BashOperator实现（实现成本更低） #

原生BashOperator自带skip_exit_code特性，你只需要把SparkSubmitOperator的配置手动拼接为spark-submit的CLI命令即可，代码示例如下：

from airflow.operators.bash import BashOperator

def spark_job(task_id: str, cfg: ArgList, main_class: str, dag: DAG, skip_exit_code: int = 10) -> BaseOperator:
    args_str = ' '.join(cfg.to_arg_list())
    # 拼接完整spark-submit命令，参数和你原有SparkSubmitOperator配置保持一致
    spark_submit_cmd = f"""
    spark-submit \
    --class {main_class} \
    --executor-cores 1 \
    --executor-memory 1g \
    --num-executors 1 \
    --total-executor-cores 2 \
    --driver-memory 1g \
    {SPARK_JOBS_JAR} {args_str}
    """
    return BashOperator(
        task_id=task_id,
        bash_command=spark_submit_cmd,
        skip_exit_code=skip_exit_code,
        dag=dag
    )

# 调用示例：指定退出码为2时触发跳过
cfg = CheckDataCfg(...)
check_data_task = spark_job('check-data', cfg, 'etljobs.spark.CheckDataRecieved', dag, skip_exit_code=2)
check_data_task >> '<待条件跳过的下游任务>'

注意：如果你的Spark集群是YARN部署模式，需要在spark-submit命令中补充--master yarn --deploy-mode cluster等对应配置，和你原生提交参数保持一致即可。

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