airflow - 如何将 Xcom 的值传递给另一个运算符？

Question

DockerOperator 有一个参数 xcom_push，设置后会将 Docker 容器的输出推送到 Xcom:

t1 = DockerOperator(task_id='run-hello-world-container',
                    image='hello-world',
                    xcom_push=True, xcom_all=True,
                    dag=dag)

在 Xcom 下的管理界面中，我可以使用键 return_value 查看这些值。但是，如何在 DAG 中访问它们？

如果我尝试:

t1_email_output = EmailOperator(task_id='t1_email_output',
                                to='user@example.com',
                                subject='Airflow sent you an email!',
                                html_content={{ ti.xcom_pull(task_ids='return_value') }},
                                dag=dag)

我得到 Broken DAG: [PATH] name 'ti' is not defined。

如果我尝试:

t1_email_output = EmailOperator(task_id='t1_email_output',
                                to='user@example.com',
                                subject='Airflow sent you an email!',
                                html_content=t1.xcom_pull(task_ids='return_value'),
                                dag=dag)

我得到 Broken DAG: [PATH] xcom_pull() missing 1 required positional argument: 'context'。

Answer 1

您需要传递从中提取 xcom 的任务 ID，而不是变量名称在您的示例中，它将是

{{ ti.xcom_pull('run-hello-world-container') }}

同样在第二个片段中它应该是 "ti" 而不是 "t1"

html_content=ti.xcom_pull('run-hello-world-container'),