我有一个由另一个dag触发的dag。我已经通过DagRunOrder().payload
字典向这个dag传递了一些配置变量,就像official example那样。
现在在这个dag中,我有另一个TriggerDagRunOperator
来启动第二个dag,并希望通过这些相同的配置变量。
我成功地访问了PythonOperator
中的负载变量,如下所示:
def run_this_func(ds, **kwargs):
print("Remotely received value of {} for message and {} for day".format(
kwargs["dag_run"].conf["message"], kwargs["dag_run"].conf["day"])
)
run_this = PythonOperator(
task_id='run_this',
provide_context=True,
python_callable=run_this_func,
dag=dag
)
但同样的模式在TriggerDagRunOperator
中不起作用:
def trigger(context, dag_run_obj, **kwargs):
dag_run_obj.payload = {
"message": kwargs["dag_run"].conf["message"],
"day": kwargs["dag_run"].conf["day"]
}
return dag_run_obj
trigger_step = TriggerDagRunOperator(
task_id="trigger_modelling",
trigger_dag_id="Dummy_Modelling",
provide_context=True,
python_callable=trigger,
dag=dag
)
它会生成有关使用provide_context
的警告:
INFO - Subtask: /usr/local/lib/python2.7/dist-packages/airflow/models.py:1927: PendingDeprecationWarning: Invalid arguments were passed to TriggerDagRunOperator. Support for passing such arguments will be dropped in Airflow 2.0. Invalid arguments were:
INFO - Subtask: *args: ()
INFO - Subtask: **kwargs: {'provide_context': True}
INFO - Subtask: category=PendingDeprecationWarning
这个错误表明我没有通过conf:
INFO - Subtask: Traceback (most recent call last):
INFO - Subtask: File "/usr/local/lib/python2.7/dist-packages/airflow/models.py", line 1374, in run
INFO - Subtask: result = task_copy.execute(context=context)
INFO - Subtask: File "/usr/local/lib/python2.7/dist-packages/airflow/operators/dagrun_operator.py", line 64, in execute
INFO - Subtask: dro = self.python_callable(context, dro)
INFO - Subtask: File "/home/user/airflow/dags/dummy_responses.py", line 28, in trigger
INFO - Subtask: "message": kwargs["dag_run"].conf["message"],
INFO - Subtask: KeyError: 'dag_run'
我尝试过的第二种模式也没有成功,它使用了params
参数,如下所示:
def trigger(context, dag_run_obj):
dag_run_obj.payload = {
"message": context['params']['message'],
"day": context['params']['day']
}
return dag_run_obj
trigger_step = TriggerDagRunOperator(
task_id="trigger_modelling",
trigger_dag_id="Dummy_Modelling",
python_callable=trigger,
params={
"message": "{{ dag_run.conf['message'] }}",
"day": "{{ dag_run.conf['day'] }}"
},
dag=dag
)
此模式不产生错误,而是将参数作为字符串传递给下一个dag,即它不计算表达式。
如何访问第二个dag的TriggerDagRunOperator
中的配置变量?
目前没有回答
相关问题 更多 >
编程相关推荐