Airflow xcom_pull 不提供相同上游任务实例运行的数据，而是提供最新数据

Question

我正在创建一个 Airflow @daily DAG，它有一个get_daily_dataBigQueryGetDataOperator的上游任务，它根据 execution_date 和下游依赖任务（PythonOperator）获取数据，通过 xcom_pull 使用以上基于日期的数据。当我运行气流回填命令时，process_data_from_bq我正在执行 xcom_pull的下游任务，它只获取最近的数据，而不是下游任务期望的同一执行日期的数据。Airfow 文档说如果我们传递如果 xcom_pull 为 task_ids 传递单个字符串，则返回来自该任务的最新 XCom 值 但是它没有说明如何获取 DAG 执行的同一实例的数据。

我经历了一个相同的问题如何在同一个 DAG 运行中从其他任务实例中提取 xcom 值（不是最近的一个）？然而，那里给出的一个解决方案是我已经在做的。但似乎它不是正确的答案。

DAG 定义：

dag = DAG(
    'daily_motor',
    default_args=default_args,
    schedule_interval='@daily'
)

#This task creates data in a BigQuery table based on execution date
extract_daily_data  = BigQueryOperator(
    task_id='daily_data_extract',
    use_legacy_sql=False,
    write_disposition='WRITE_TRUNCATE',
    allow_large_results=True,
    sql=policy_by_transaction_date_sql('{{ ds }}'), 
    destination_dataset_table='Test.daily_data_tmp',
    dag=dag)


get_daily_data = BigQueryGetDataOperator(
    task_id='get_daily_data',
    dataset_id='Test',
    table_id='daily_data_tmp',
    max_results='10000',
    dag=dag

)


#This is where I need to pull the data of the same execution date/same instance of DAG run not the most recent task run

def process_bq_data(**kwargs):
    bq_data = kwargs['ti'].xcom_pull(task_ids = 'get_daily_data')
    #This bq_data is most recent one not of the same execution date
    obj_creator = IibListToObject()
    items = obj_creator.create(bq_data, 'daily')
    save_daily_date_wise(items)


process_data = PythonOperator(
    task_id='process_data_from_bq',
    python_callable=process_bq_data,
    provide_context=True,
    dag = dag
)

get_daily_data.set_upstream(extract_daily_data)
process_data.set_upstream(get_daily_data)

Answer 1

您必须收到最新的 Xcom 值。您还需要确保值实际上来自与预期相同的execution_date：

:param include_prior_dates: 

如果为 False，则仅返回当前execution_date 中的 XCom。如果为 True，则还会返回之前日期的 XCom。