我有一个要在 DAG_1 和 DAG_2 中使用的 task_a。这在气流中可能吗?
task_a = SomeOperator(
task_id='some_id',
bash_command='some_command',
#instead of just
dag= DAG_1 # I want to assign this task to multiple dags
#dag=assign_multiple_dags_here(DAG_1 and DAG_2)
)
Run Code Online (Sandbox Code Playgroud)
这可能吗?
你总是可以做一些事情,partial
然后将它分配给 2 个不同的 dag:
from functools import partial
task_template = partial(SomeOperator, some_id='id', some_command='cmd')
task_template(dag=dag1)
task_template(dag=dag2)
Run Code Online (Sandbox Code Playgroud)
您也可以创建一个执行此操作的函数:
def create_task(dag):
return SomeOperator(some_id='id', some_command='cmd', dag=dag)
for d in (dag1, dag2):
create_task(d)
Run Code Online (Sandbox Code Playgroud)