Airflow中SparkKubernetesOperator任务失败后K8s Pod未删除如何解决？

默认情况下，SparkKubernetesOperator仅负责提交SparkApplication资源到Kubernetes，不会在Airflow任务被标记为失败时自动清理对应的K8s资源（包括Pod）。可以通过以下几种方式解决：

方法一：添加任务失败回调函数，主动删除SparkApplication资源 #

编写回调函数，在任务失败时调用Kubernetes API删除对应的SparkApplication，触发Pod的清理流程。

修改你的代码如下：

from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
from airflow.exceptions import AirflowException
from datetime import datetime

def delete_spark_app_on_failure(context):
    # 从XCom中获取提交的SparkApplication名称
    spark_app_name = context['task_instance'].xcom_pull(task_ids='sample-name')['metadata']['name']
    namespace = "batch"
    
    # 初始化Kubernetes Hook
    hook = KubernetesHook(conn_id="kubernetes_default")
    api_client = hook.get_api_client()
    
    # 调用自定义资源API删除SparkApplication
    from kubernetes.client import CustomObjectsApi
    custom_api = CustomObjectsApi(api_client)
    try:
        custom_api.delete_namespaced_custom_object(
            group="sparkoperator.k8s.io",
            version="v1beta2",
            namespace=namespace,
            plural="sparkapplications",
            name=spark_app_name
        )
        print(f"已成功删除SparkApplication: {spark_app_name}")
    except Exception as e:
        raise AirflowException(f"删除SparkApplication失败: {str(e)}")

dag = DAG(
    "dag-name", 
    default_args=default_args,
    description='submit dag',
    schedule_interval="30 1 * * *",
    start_date = datetime(2022, 7, 26),
    )

t1 = SparkKubernetesOperator(
        task_id='sample-name',
        namespace="batch",
        application_file="k8s/sample.yaml",
        do_xcom_push=True,
        dag=dag,
        params = {"processDate": process_date},
        # 绑定失败回调函数
        on_failure_callback=delete_spark_app_on_failure
    )

t1_sensor = SparkKubernetesSensor(
        task_id='sample-monitor',
        namespace="batch",
        application_name="{{ task_instance.xcom_pull(task_ids='sample-name')['metadata']['name'] }}",
        kubernetes_conn_id="kubernetes_default",
        dag=dag,
    )

t1 >> t1_sensor

注意：需根据你实际使用的Spark Operator版本调整group和version参数（部分环境使用spark.apache.org作为group，版本为v1beta1）。

方法二：在SparkApplication YAML中配置TTL自动清理 #

在k8s/sample.yaml里添加TTL控制器配置，让Kubernetes在SparkApplication完成或失败后自动清理资源，无需依赖Airflow的回调：

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: sample-app
spec:
  # 原有业务配置...
  ttlSecondsAfterFinished: 3600 # 任务结束后1小时自动删除资源及关联Pod

这种方式是Kubernetes层面的自动清理，不管Airflow任务状态如何，只要SparkApplication进入终止状态，就会在指定时间后被清理。

方法三：结合重试逻辑添加前置清理 #

如果任务配置了重试，可以在default_args中添加on_retry_callback，在重试前清理上一次提交的SparkApplication，避免残留Pod。

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