Блог

У меня возникли проблемы с доступом к значениям xcom воздушного потока из субдап.Я не получаю ничего в значениях Текущий дизайн включает в себя конвейер данных оркестровки, который состоит из главной группы доступности данных, которая запускает около 16 подзаголовков потока данных с помощью TriggerDagRunOperator . Master dag и другие дочерние группы-это отдельные файлы dag python, запущенные в Google Composer. Я создал отдельный xcomutil.py файл, который имеет функцию python для xcom

 def xpush(**kwargs): ti = kwargs['ti'] x_push = ti.xcom_push(key=kwargs['key'], value=kwargs['value']) print(x_push)   def xpull(**kwargs): ti = kwargs['ti'] x_pull = ti.xcom_pull(dag_id='d1', key=kwargs['key']) return x_pull  

Из мастера dag я создал pythonoperator

 pushtask = PythonOperator(  task_id='pushtask',  python_callable=xcomutil.x_push,  op_kwargs={'key': master_parent, 'value': run_id},  provide_context=True  

Из одного из поддагов я создал еще один оператор python для извлечения xcom

 pulltask = PythonOperator(  task_id='pulltask',  python_callable=xcomutil.x_pull,  op_kwargs={'key': master_parent},  provide_context=True  )  

Я хочу извлечь значение ключа главной группы dag в вспомогательной группе dag, которое хранится как ключ : значение master_parent : руководство__2021-00-20T07:00:54.040052 00:00 в таблице xcom

Но я не получаю никакой ценности . Также я пытаюсь передать parent_dag_id = xcom_utility_v2.x_pull в задание потока данных. Можно ли получить к нему доступ, вызвав return x_pull из файла утилиты xcom?

По какой-то причине экземпляр задачи из subdag передается как dagid, и я думаю, что это может быть причиной отсутствия значения