从 sub dag 中拉取 xcom

Question

我正在使用一个主 dag (main_dag)，其中包含多个子标签，每个子标签都有多个任务。我从 subdagA taskA 中推送了一个 xcom，但我在 subdagB taskB 中拉动了那个 xcom。由于 xcom_pull() 中的 dag_id 参数默认为 self.dag_id 我一直无法提取必要的 xcom。我想知道如果有更好的方法来设置这种情况，我将如何做到这一点 and/or，这样我就不必处理这个问题了。

我目前在 subdagB 中所做的示例：

def subdagB(parent_dag, child_dag, start_date, schedule_interval):
    subdagB = DAG('%s.%s' % (parent_dag, child_dag), start_date=start_date, schedule_interval=schedule_interval)
    start = DummyOperator(
        task_id='taskA',
        dag=subdagB)
    tag_db_template = '''echo {{ task_instance.xcom_pull(dag_id='dag.main_dag.subdagA', task_ids='taskA') }};'''
    t1 = BashOperator(
        task_id='taskB',
        bash_command=tag_db_template,
        xcom_push=True,
        dag=subdagB)
    end = DummyOperator(
        task_id='taskC',
        dag=subdagB)
    t0.set_upstream(start)
    t1.set_upstream(t0)
    end.set_upstream(t1)
    return subdagB

提前感谢您的帮助！

Answer 1

只要你覆盖

中的dag_id就应该没问题

[Operator].xcom_pull(dag_id=dag_id, ...) 或
[TaskInstance].xcom_pull(dag_id=dag_id, ...)

只需确保

dag_id = "{parent_dag_id}.{child_dag_id}"

如果你能让你的示例更完整，我可以在本地尝试运行，但我测试了一个（类似的）示例并且 cross-subdag xcoms 按预期工作.

从 sub dag 中拉取 xcom

Pulling xcom from sub dag

python

airflow