为python列表中存在的气流任务创建动态工作流程

Question

我有以下方式的列表 -

[['x_api'，'y_api'，....]，['z_api'，'p_api'，...]，[....]，[。 ..] ....]

在这里，每个API名称对应于Pythonoperator。

在气流中，我想创建任务依赖项，以便从起始虚拟任务中，我应该对主列表中的每个列表中的每个列表进行并行任务，并且列表列表中的操作员应按顺序执行：

我该怎么做？感谢您的任何帮助！

现有代码：

    args = {
            'depends_on_past': False,
            'start_date': datetime.now(),
            'email': '',
            'email_on_failure': False,
            'email_on_retry': False,
            'retries': 3,
            'retry_delay': timedelta(minutes=1)
        }   
    
    dag = DAG(dag_id, default_args=args, schedule_interval=None)
        
    with dag:
        tasks = []
        tmp, tmp2 = set(), set()
        Start = DummyOperator(
                task_id='Start',
                dag=dag
        )
    
        End = DummyOperator(
                task_id='End',
                dag=dag
        )
    
        for i in dags:
            for j in i:
                if 'APIs' in list(i.keys()):
                        
                    for l in i['APIs']:
                            tab = DummyOperator(
                            task_id=l['api'] + "_API",
                            dag=dag
                            )
                            tmp.add(tab)    
                elif 'tables' in list(i.keys()):
                        
                    for k in i['tables']:
                        tab2 = DummyOperator(
                            task_id=k['table'] + "_API",
                            dag=dag
                        )
                        tmp2.add(tab2)

      tasks.append(list(tmp))
      tasks.append(list(tmp2))

      for task in tasks:
        for op in range(0, len(task)-1):
            Start.set_downstream(task[op])
            task[op].set_downstream(task[op+1])
            task[op+1].set_downstream(End)

Answer 1

这是我想到的解决方案 -

    with dag:
Start = ShortCircuitOperator(
    task_id = 'Create_Postgres_Schema',
    python_callable = create_postgres_schema,
    ignore_downstream_trigger_rules = True
)

End = DummyOperator (
    task_id = 'End_of_Data_extraction',
    trigger_rule = TriggerRule.ALL_DONE
)

# Get list of first and last task. This is done to enable the task flow in Airflow
start_task_list = list(map(itemgetter(0), tasks_master_list))
end_tasks_list = list(map(itemgetter(-1), tasks_master_list))

Start >> start_task_list

for task in tasks_master_list:
    for op in range(len(tasks_master_list)-1):
        task[op] >> task[op+1]

end_tasks_list >> End