Airflow 2.4.3：从PythonOperator获取BigQueryOperator任务信息遇阻

先处理方案2的TypeError错误 #

这个错误是因为你的PythonOperator任务函数未接收context参数，或未开启context传递。两种解决方式：

• 在函数定义中添加**kwargs或context参数，示例：

  def task3_function(**kwargs):
      # 函数逻辑
  

• 定义PythonOperator时设置provide_context=True（Airflow 2.x仍支持该参数）

完整获取前序任务元数据的方案 #

要获取Task1（BigQueryOperator）和Task2（PythonOperator）的job_id、task_id、run_id、状态、任务URL，按以下步骤实现：

1. 在Task3函数中获取前序任务的Task Instance对象 #

通过context拿到当前DAG Run，再通过DAG Run查询指定task_id的Task Instance：

from airflow.models import TaskInstance
from airflow.utils.session import create_session

def task3_function(**kwargs):
    dag_run = kwargs['dag_run']
    run_id = dag_run.run_id
    
    # 获取Task1的Task Instance
    with create_session() as session:
        ti_task1 = TaskInstance.find(dag_id=dag_run.dag_id, task_id='task1', run_id=run_id, session=session)[0]
    # 获取Task2的Task Instance
    with create_session() as session:
        ti_task2 = TaskInstance.find(dag_id=dag_run.dag_id, task_id='task2', run_id=run_id, session=session)[0]

2. 提取Task Instance的基础元数据 #

从Task Instance对象可直接获取以下信息：

• task_id: ti_task1.task_id
• run_id: ti_task1.run_id
• 任务状态: ti_task1.state
• 任务URL: ti_task1.get_url()（返回Airflow Web UI中该任务实例的完整URL）

3. 获取BigQueryOperator的job_id #

BigQueryOperator执行完成后，会自动将job_id推送到XCom（key为job_id），通过xcom_pull获取：

# 获取Task1的BigQuery job_id
task1_job_id = kwargs['ti'].xcom_pull(task_ids='task1', key='job_id')

4. 整合信息并使用 #

将所有信息整合后即可在Task3中处理，示例：

def task3_function(**kwargs):
    dag_run = kwargs['dag_run']
    run_id = dag_run.run_id
    ti = kwargs['ti']
    
    # 获取Task1的Task Instance和job_id
    with create_session() as session:
        ti_task1 = TaskInstance.find(dag_id=dag_run.dag_id, task_id='task1', run_id=run_id, session=session)[0]
    task1_job_id = ti.xcom_pull(task_ids='task1', key='job_id')
    
    # 获取Task2的Task Instance和job_id（PythonOperator的job_id即Task Instance的job_id）
    with create_session() as session:
        ti_task2 = TaskInstance.find(dag_id=dag_run.dag_id, task_id='task2', run_id=run_id, session=session)[0]
    task2_job_id = ti_task2.job_id
    
    # 输出或处理信息
    print(f"Task1详情: task_id={ti_task1.task_id}, run_id={ti_task1.run_id}, state={ti_task1.state}, job_id={task1_job_id}, url={ti_task1.get_url()}")
    print(f"Task2详情: task_id={ti_task2.task_id}, run_id={ti_task2.run_id}, state={ti_task2.state}, job_id={task2_job_id}, url={ti_task2.get_url()}")

5. 正确定义Task3的PythonOperator #

确保开启context传递，示例：

task3 = PythonOperator(
    task_id='task3',
    python_callable=task3_function,
    provide_context=True,
    dag=dag
)

方案1失效的原因 #

• PythonOperator的状态、task_id等元数据无需手动XCom推送，Task Instance本身已存储这些信息，直接查询更可靠。
• BigQueryOperator仅自动推送job_id到XCom，其余元数据均存储在对应的Task Instance中，需结合Task Instance对象获取。

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