Airflow动态任务依赖仅对列表最后一个元素生效问题排查

• 依赖配置语句get_parameters >> feed_check >> feed_copy放在了双层for循环的外部，循环运行过程中feed_check和feed_copy变量会被每次迭代生成的新任务反复覆盖，等循环完全执行结束后，这两个变量仅保存了最后一次迭代生成的任务对象，因此仅最后一组任务被正确配置了依赖关系，此前生成的其他任务都没有绑定依赖。

将依赖配置语句移动到双层for循环的内部，每次生成对应feed_check和feed_copy任务后立即配置依赖，即可实现所有任务对的依赖正常绑定。

EXPORT_BUCKET='gs://export'
SOURCE_BUCKET='gs://source'
system_list=["main"]
type_list=["road_cell_counters", "road_counters", "impact_counters", "cell_counters"]

...

with models.DAG(dag_id='return_feed',
        schedule_interval=None,
        description='return feed',
        default_args=default_dag_args) as dag: 

...
 
        for system in system_list:
            for type in type_list:
                feed_check = BashOperator(
                task_id=str(type)+'_check_'+str(system),
                project_id=project,
                xcom_push=True,
                bash_command='''gsutil -q stat {source_bucket}/system={system}/type={type}/year={{ ti.xcom_pull(key="YEAR") }}/month={{ ti.xcom_pull(key="MONTH") }}/day={{ ti.xcom_pull(key="DAY") }}/_SUCCESS'''.format(source_bucket=SOURCE_BUCKET, system=system, type=type),
                dag = dag
                )        

                feed_copy = BashOperator(
                task_id=str(type)+'_copy_'+str(system),
                project_id=project,
                bash_command='''gsutil cp -r {source_bucket}/system={system}/type={type}/year={{ ti.xcom_pull(key="YEAR") }}/month={{ ti.xcom_pull(key="MONTH") }}/day={{ ti.xcom_pull(key="DAY") }} {export_bucket}/system={system}/type={type}/year={{ ti.xcom_pull(key="YEAR") }}/month={{ ti.xcom_pull(key="MONTH") }}/'''.format(source_bucket=SOURCE_BUCKET, export_bucket=EXPORT_BUCKET, system=system, type=type),
                dag = dag
                )
                # 依赖配置移到循环内部
                get_parameters >> feed_check >> feed_copy

内容的提问来源于stack exchange，提问作者Aleksander Lipka