从 Airflow 中的所有任务中获取带有密钥的 xcoms答案

【问题标题】：Get xcoms with a key from all tasks in Airflow从 Airflow 中的所有任务中获取带有密钥的 xcoms
【发布时间】：2021-12-11 07:08:04
【问题描述】：

我有两个 Airflow 任务使用相同的键 srcDbName 推送 xcoms，但具有不同的值。这两个任务之后是一个任务，它使用键 srcDbName 读取 xcoms 并打印它们的值。请看下面的代码：

def _fill_facebook_task(ti):
    ti.xcom_push(key='srcDbName', value='SRC_PL_Facebook')

def _fill_trip_advisor_task(ti):
    ti.xcom_push(key='srcDbName', value='SRC_PL_TripAdvisor')

def _pm_task(ti):
    values = ti.xcom_pull(key='srcDbName')
    print(', '.join(values))

facebook = PythonOperator(
    task_id="fill-facebook",
    python_callable= _fill_facebook_task,
    dag=dag
)

tripAdvisor = PythonOperator(
    task_id="fill-trip-advisor",
    python_callable=_fill_trip_advisor_task,
    dag=dag
)

pm = PythonOperator(
    task_id="premises-matching",
    python_callable=_pm_task,
    dag=dag
)

facebook  >> pm
tripAdvisor >> pm

我希望 pm 任务应该打印

SRC_PL_Facebook、SRC_PL_TripAdvisor

（或以不同的顺序）因为documentation for xcom_pull 声明：

:param task_ids: 只有来自具有匹配 id 的任务的 XComs 才会被拉。可以通过 None 来移除过滤器。

其实是打印出来的

S、R、C、_、P、L、_、F、a、c、e、b、o、o、k

是否可以使用给定密钥从所有上游任务中读取所有xcoms？

【问题讨论】：

标签： airflow

【解决方案1】：

要读取所有 xcom，您需要将所有上游 task_instance 名称作为参数传递给 xcom_pull。 xcom_pull 方法的文档肯定没有说清楚——它只是说：

If a single task_id string is provided, the result is the value of the most
recent matching XCom from that task_id. If multiple task_ids are provided, a
tuple of matching values is returned. None is returned whenever no matches
are found.

但它还应该提到，如果您不传递任何 task_ids，那么xcom_pull 将只返回它找到的第一个（如果有的话）匹配值。您可以在code for airflow.models.taskinstance 中验证该行为。

【讨论】：

感谢源代码链接，很有帮助！