Airflow 2.3.3中TaskGroup内SparkKubernetesOperator与Sensor依赖报错

在Airflow的TaskGroup中，任务的完整task_id格式为TaskGroupID.子任务task_id。你当前的Sensor任务中，xcom_pull仅指定了子任务的task_id（task-1），没有带上TaskGroup的前缀（tg-task-1），导致无法正确拉取到上游任务的XCom数据，返回None，进而触发'None' has no attribute 'metadata'的报错。

你需要在xcom_pull中指定包含TaskGroup前缀的完整task_id，或者通过task_group_id参数明确指定任务所在的TaskGroup，以下是几种可行的修改方式：

方式1：直接使用完整task_id #

修改Sensor的application_name参数，将task_id改为tg-task-1.task-1：

task_1_sensor = SparkKubernetesSensor(
    task_id='task-1-sensor',
    namespace="batch",
    application_name="{{ task_instance.xcom_pull(task_ids='tg-task-1.task-1')['metadata']['name'] }}",
    kubernetes_conn_id="kubernetes_default",
    dag=dag,
    attach_log=True,
)

方式2：通过task_group_id参数指定 #

利用Airflow的xcom_pull支持的task_group_id参数，无需硬编码完整task_id：

task_1_sensor = SparkKubernetesSensor(
    task_id='task-1-sensor',
    namespace="batch",
    application_name="{{ task_instance.xcom_pull(task_ids='task-1', task_group_id='tg-task-1')['metadata']['name'] }}",
    kubernetes_conn_id="kubernetes_default",
    dag=dag,
    attach_log=True,
)

方式3：动态获取当前TaskGroup ID（推荐） #

通过当前任务的task_group_id变量动态关联，避免后续修改TaskGroup名称时需要同步修改代码：

task_1_sensor = SparkKubernetesSensor(
    task_id='task-1-sensor',
    namespace="batch",
    application_name="{{ task_instance.xcom_pull(task_ids='task-1', task_group_id=task.task_group_id)['metadata']['name'] }}",
    kubernetes_conn_id="kubernetes_default",
    dag=dag,
    attach_log=True,
)

with TaskGroup("tg-task-1", default_args=default_args) as tg_task_1:
    task_1 = SparkKubernetesOperator(
                task_id='task-1',
                namespace="batch",
                application_file="k8s/task-1.yaml",
                do_xcom_push=True,
                dag=dag,
            )

    task_1_sensor = SparkKubernetesSensor(
        task_id='task-1-sensor',
        namespace="batch",
        application_name="{{ task_instance.xcom_pull(task_ids='task-1', task_group_id=task.task_group_id)['metadata']['name'] }}",
        kubernetes_conn_id="kubernetes_default",
        dag=dag,
        attach_log=True,
    )

task_1 >> task_1_sensor

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