Airflow - 将 Xcom Pull 结果传递给 TriggerDagRunOperator conf

Question

有谁知道下面的代码有什么问题：

本质上，我正在调用 TriggerDagRunOperator，并且我正在尝试基于 XCOM Pull 将一些 conf 传递给它。

触发 DAG：

def _should_trigger(**_):
    return {'Message': 'Hello World'}


should_trigger = PythonOperator(
    task_id="should_trigger",
    python_callable=_should_trigger,
    provide_context=True,
)

trigger_bar_dag = TriggerDagRunOperator(
    task_id="trigger_bar_dag",
    trigger_dag_id="bar",
    conf={"payload": "{{ task_instance.xcom_pull('should_trigger') }}"},
)

目标 DAG：

@dag(dag_id="bar",
     default_args=default_args,
     schedule_interval=None
     )
def tasks():
    run_this = PythonOperator(
        task_id="run_this",
        python_callable=run_this_func,
        provide_context=True)

由于某种原因，在 run_this_func 中，我得到了有效载荷：无。我似乎无法通过来自 xcom 拉取的 conf 流传递值。有谁知道如何做到这一点。我还尝试了不同的 xcom pull 变体，例如： ti.xcom_pull(key='return_value', task_ids=['should_trigger']) 无济于事。

谢谢，

Answer 1

你需要指定should_trigger >> trigger_bar_dag，否则XCom记录可能还不存在，你会得到：

[2021-06-06 08:23:35,898] {logging_mixin.py:104} INFO - {'payload': 'None'}

但是一旦我添加了这个关系，我就会得到：

[2021-06-06 08:21:41,356] {logging_mixin.py:104} INFO - {'payload': "{'Message': 'Hello World'}"}

与

def run_this_func(**context):
    print(context['params'])