Considering a parent dag that pushes value to xcom, how to retrieve dag from subdag?
What I've tried:
#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
And the child subdag:
#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
Instead of child subdag to log and return the [1,2]
, it returned string '{{ ti.xcom_pull(task_ids='push_value', dag_id='MyParentDag' }}'
{{ti.xcom_pull(task_ids='push_value', dag_id='" + PARENT_DAG_NAME + "')}}
In the above code a parenthesis is missing
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With