Airflow：如何从父DAG传递xcom到子DAG

Question

考虑到一个父DAG向XCom推送值，如何从子DAG中检索DAG？

我尝试过的方法：

#parent_dag.py

PARENT_DAG_NAME = "MyParentDag"
CHILD_DAG_NAME = "MyChildDag"

main_dag = DAG(
  dag_id=PARENT_DAG_NAME,
  schedule_interval="@hourly",
  start_date=DAG_START_DATE
)


def push_value(**kwargs):
    ''' push into Xcom '''
    return [1, 2]

t1 = PythonOperator(task_id='push_value',
                       python_callable=push_value,
                       retries=3,
                       dag=main_dag)

subdag_1 = SubDagOperator(
  subdag=Sub_Dag1(
      PARENT_DAG_NAME,
      CHILD_DAG_NAME,
      main_dag.start_date,
      main_dag.schedule_interval,
      "'{{ ti.xcom_pull(task_ids='push_value', dag_id='" + PARENT_DAG_NAME + "' }}'"
  ),
  task_id=CHILD_DAG_NAME,
  dag=main_dag,
)
t1 >> subdag_1

还有子DAG：

#subdag1.py


def use_pushed_val(pushed_val, ds, **kwargs):
    log.info(pushed_val)
    return pushed_val

def Sub_Dag1(parent_dag_name, child_dag_name, start_date, schedule_interval, pushed_val):
  dag = DAG(
    '%s.%s' % (parent_dag_name, child_dag_name),
    schedule_interval=schedule_interval,
    start_date=start_date,
  )

  childTask = PythonOperator(
      task_id='child_task',
      python_callable=use_pushed_val,
      op_kwargs = {'pushed_val' : pushed_val},
      provide_context=True,
      dag=dag
    )

return dag

将子DAG记录并返回[1,2]的操作失败，它返回了字符串'{{ ti.xcom_pull(task_ids='push_value', dag_id='MyParentDag' }}'

- DougKruger

---

你能找到从DAG传递值到子DAG的方法吗？ - Gagan

这是一个可行的解决方案，只需添加一个缺失的括号即可。 - 42n4

Answer 1

{{ti.xcom_pull(task_ids='push_value', dag_id='" + PARENT_DAG_NAME + "')}}

在上面的代码中，缺少一个括号。

- RaHuL KuMaR ShArMa

---

是的，它解决了问题。 - 42n4

Answer 2

我看到你已经设置了provide_context=True，这很好。这是我使用context参数在父/子DAG之间传递变量的方法。

def push_value(**context):
    context['ti'].xcom_push(key='my_key', value='my_value')

def use_pushed_val(**context):
    value_from_parent = context['ti'].xcom_pull(task_ids=t1.task_id, key='my_key')

- Thinh Pham

---

必须将dag_id='master_id'添加到xcom_pull中。 - 42n4