如何在TaskGroup类实例中从TaskFlow任务向Sensor传递XCom

问题描述 #

我在Airflow中创建了MyTaskGroup类，用于在DAG中多次实例化通用任务流程。类内包含一个TaskFlow任务refresh_schema_connections，负责提交API作业请求并返回响应，需要从中提取job_id传递给MyJobSensor以轮询作业完成状态，但始终无法正确编写XCom拉取语法。

我明确问题在于拉取XCom时需要为task_id添加当前实例的group_id前缀，但每个类实例的group_id各不相同，需要修正语法让它自动获取当前task_instance的group_id。

代码示例 #

class MyTaskGroup(TaskGroup):
    def __init__(self, database_connection):
        super().__init__(group_id=f'register_schemas{database_connection['name']}', tooltip=f'Registering Schemas')

        @task(task_group=self)
        def refresh_schema_connections(database_connection):
            client = get_api_client()
            return client.submit_job(database_connection)
        
        schema_job = refresh_schema_connections(database_connection['id'])
        
        wait_for_schema_job_completion = MyJobSensor(
            task_id=f"wait_for_schema_job_completion", 
            task_group=self,
            mode="reschedule",
            job_id="{{ task_instance.xcom_pull(task_instance.task_group_id + '.refresh_schema_connections')['output']['id'] }}"
        )
        schema_job >> wait_for_schema_job_completion

尝试过的错误写法 #

• job_id=schema_job.output['id']
• job_id="{{ task_instance.xcom_pull(task_instance.group_id + '.refresh_schema_connections')['output']['id'] }}"

解决方案 #

方案1：使用TaskFlow API直接引用输出（推荐） #

Airflow 2.x的TaskFlow API支持直接通过任务对象的output属性获取返回值，无需手动编写XCom拉取模板。你之前的写法错误是因为少了一层output键，修正后即可正常获取job_id：

wait_for_schema_job_completion = MyJobSensor(
    task_id="wait_for_schema_job_completion", 
    task_group=self,
    mode="reschedule",
    job_id=schema_job.output['output']['id']
)

这种方式不需要关心group_id的拼接逻辑，Airflow会自动处理跨TaskGroup的XCom传递，代码更简洁易维护。

方案2：修正Jinja模板的XCom拉取语法 #

如果必须使用Jinja模板方式，需要注意两个关键修正点：

1. task_instance的正确属性为task_group_id，而非group_id
2. xcom_pull需要明确指定task_id参数

修正后的代码如下：

wait_for_schema_job_completion = MyJobSensor(
    task_id="wait_for_schema_job_completion", 
    task_group=self,
    mode="reschedule",
    job_id="{{ task_instance.xcom_pull(task_id=task_instance.task_group_id + '.refresh_schema_connections')['output']['id'] }}"
)

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