如何在Airflow中使用SSHOperator终止运行在YARN上的spark-submit作业

• Local模式下Spark作业的全量逻辑都运行在spark-submit进程内，Airflow触发kill时，SSHOperator终止远程SSH会话对应的spark-submit进程即可直接停止作业，因此可以正常生效。
• YARN模式下spark-submit仅作为客户端进程存在，作业的Driver和Executor都运行在YARN集群的资源容器中。仅终止spark-submit进程不会影响YARN上运行的作业，因此会出现UI kill不生效的问题。

方案1：适配现有SSHOperator逻辑 #

1. 给SSHOperator添加get_pty=True参数，开启伪终端分配，确保kill操作可以终止该SSH会话下的所有关联子进程。
   修正后的基础代码参考：

task1 = SSHOperator(
    ssh_hook=sshHook,
    task_id='test_spark',
    do_xcom_push=False,
    command="你的提交命令",
    get_pty=True
)

注意你提供的示例代码中task_id='test_spark缺少闭合单引号，运行前需要修正。
2. 将spark-submit命令封装为带信号处理的shell脚本，捕获Airflow kill时发送的SIGTERM信号，主动调用YARN命令终止对应作业，脚本示例如下：

#!/bin/bash
# 提交Spark作业，捕获YARN application ID
APP_ID=$(spark-submit \
  --master yarn \
  --deploy-mode cluster \
  # 替换为你自己的其他Spark提交参数 \
  2>&1 | grep -o 'application_[0-9]\+_[0-9]\+')

# 注册信号回调，收到SIGTERM时终止YARN作业
trap "yarn application -kill $APP_ID" SIGTERM

# 等待作业运行结束
wait

确保执行该脚本的服务器已配置YARN客户端环境，且执行用户拥有对应YARN作业的终止权限，再将SSHOperator的command参数改为执行上述脚本即可。

方案2：替换为官方SparkSubmitOperator #

Airflow官方提供的SparkSubmitOperator原生适配YARN作业生命周期管理，任务被kill时会自动调用YARN接口终止对应的Application，无需手动封装信号处理逻辑，适配性更高，配置示例如下：

from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator

task1 = SparkSubmitOperator(
    task_id='test_spark',
    conn_id='spark_default', # 提前在Airflow连接中配置好Spark连接信息
    application="你的作业jar包/py文件路径",
    master="yarn",
    deploy_mode="cluster",
    # 其他自定义Spark配置参数
)

内容的提问来源于stack exchange，提问作者Hiệp