Airflow - xcom 值访问自定义操作符

Question

自过去 6 个月以来，我一直在使用 Airlfow。我很高兴在 Airflow 中定义工作流程。
我有以下场景，我无法获得 xcom 值(以黄色突出显示)。

请在以下示例代码中找到代码:

工作流程

def push_function(**context):
context['ti'].xcom_push(key='reportid', value='xyz')

dummy_operator = DummyOperator(
task_id='Start',
dag=main_dag
)

push_function_task = PythonOperator(
    task_id='push_function',
    provide_context=True,
    python_callable=push_function,
    op_kwargs={},
    dag=main_dag)


push_function_task .set_upstream(dummy_operator)

custom_task = CustomOperator(
        dag=main_dag,
        task_id='import_data',
        provide_context=True,
        url="http://www.google.com/{}".format("{{task_instance.xcom_pull(task_ids='push_function')}}")

     )

custom_task .set_upstream(push_function_task)

备注 :
1. CustomOperator 是我自己的操作符，用于下载给定 URL 的数据

请帮我。

谢谢，
萨曼思

Answer 1

我相信您在推拉 XCom 时按键不匹配。每个 XCom 值都与 DAG ID、任务 ID 和 key 相关联。如果您正在插入 report_id关键，那么你也需要用它来拉。

请注意，如果未将 key 指定给 xcom_pull() ，它使用默认值 return_value .这是因为如果一个任务返回一个结果，Airflow 会自动将其推送到 return_value 下的 XCom。 key 。

这为您提供了两种解决问题的选项:

1)继续推送到report_id key ，并确保你也从中拉出

def push_function(**context):
    context['ti'].xcom_push(key='reportid', value='xyz')

...

custom_task = CustomOperator(
    ...
    url="http://www.google.com/{}".format("{{ task_instance.xcom_pull(task_ids='push_function', key='reportid') }}")
)

2) 有 push_function()将您要推送到 XCom 的值返回，然后从默认 key 中提取。

def push_function(**context):
    return 'xyz'

...

custom_task = CustomOperator(
    ...
    url="http://www.google.com/{}".format("{{ task_instance.xcom_pull(task_ids='push_function') }}")
)