Airflow DAG Loop - 如何使每次迭代顺序而不是并行
Airflow DAG Loop - How to make each iteration sequential instead of parallel
我有一个像这样的 Apache Airflow DAG:
DAG_NAME='my_dag'
sections = ["0", "1", "2", "3"]
with DAG(DAG_NAME, default_args=default_args, schedule_interval=None) as dag:
for s in sections:
a = DummyOperator(task_id=f"section_{s}_start")
b = SubDagOperator(task_id=f"init_{s}_subdag",subdag=init_section(DAG_NAME,f"init_{s}_subdag", default_args))
c = SubDagOperator(task_id=f"process_{s}_subdag", subdag=process_section(DAG_NAME,f"process_{s}_subdag", default_args))
d = SubDagOperator(task_id=f"update_{s}_subdag", subdag=update_section(DAG_NAME,f"update_{s}_subdag", default_args))
e = DummyOperator(task_id=f"section_{s}_end")
a>>b>>c>>d>>e
这段代码会像这样呈现我的任务
如何使任务顺序为:
section_0_start>>init_0_subdag>>process_0_subdag>>update_0_subdag>>section_0_end
section_0_end>>section_1_start
section_1_start>>init_1_subdag>>process_1_subdag>>update_1_subdag>>section_1_end
.....
依此类推,从第 0 节开始,以第 3 节任务结束
谢谢
像这样修改for-loop:
previous_e = None
for s in sections:
a = ...
...
e = ...
if previous_e:
previous_e >> a
a>>b>>c>>d>>e
previous_e = e
我有一个像这样的 Apache Airflow DAG:
DAG_NAME='my_dag'
sections = ["0", "1", "2", "3"]
with DAG(DAG_NAME, default_args=default_args, schedule_interval=None) as dag:
for s in sections:
a = DummyOperator(task_id=f"section_{s}_start")
b = SubDagOperator(task_id=f"init_{s}_subdag",subdag=init_section(DAG_NAME,f"init_{s}_subdag", default_args))
c = SubDagOperator(task_id=f"process_{s}_subdag", subdag=process_section(DAG_NAME,f"process_{s}_subdag", default_args))
d = SubDagOperator(task_id=f"update_{s}_subdag", subdag=update_section(DAG_NAME,f"update_{s}_subdag", default_args))
e = DummyOperator(task_id=f"section_{s}_end")
a>>b>>c>>d>>e
这段代码会像这样呈现我的任务
如何使任务顺序为:
section_0_start>>init_0_subdag>>process_0_subdag>>update_0_subdag>>section_0_end section_0_end>>section_1_start section_1_start>>init_1_subdag>>process_1_subdag>>update_1_subdag>>section_1_end
.....
依此类推,从第 0 节开始,以第 3 节任务结束
谢谢
像这样修改for-loop:
previous_e = None
for s in sections:
a = ...
...
e = ...
if previous_e:
previous_e >> a
a>>b>>c>>d>>e
previous_e = e