Airflow DAG循环任务设置依赖、创建动态任务及匹配CSV文件问题

问题1：Airflow动态任务创建与跨任务依赖设置 #

你当前的问题根源是循环生成任务时，insert_data变量被反复覆盖，最终[insert_data]列表里仅保存了最后一个（对应3.csv）的任务对象，所以才会出现仅等最后一个任务跑完就触发后续任务的问题。
正确实现逻辑如下：

1. 循环生成动态任务前先初始化一个空的任务存储列表
2. 每次生成一个Python任务就追加到该列表中
3. 最终将完整的任务列表作为上游，绑定到最终的Postgres任务
   修正后的代码片段：

# 初始化空列表存储所有动态生成的插入任务
insert_tasks = []

files = ['1.csv', '2.csv', '3.csv']

for listing in files:
    insert_data = PythonOperator(
        task_id=f'insert_data_func_{listing}',
        python_callable=insert_data_func,
        op_kwargs={'filename': listing},
        provide_context=True,
        dag=dag
    )
    # 绑定上游任务：建表任务先跑，再跑当前插入任务
    create_psql_table >> insert_data
    # 把当前插入任务追加到列表
    insert_tasks.append(insert_data)

# 所有插入任务全部跑完再跑建事实表任务
insert_tasks >> create_fact_table

问题2：自动获取指定规则的CSV文件列表 #

你已经导入了glob库，直接用它的通配符匹配功能就能实现需求，不需要手动指定文件名。
示例代码如下，假设你要获取./dags/data/目录下所有以data_开头的csv文件：

import glob
import os

# 通配符*匹配任意字符，这里匹配所有以data_开头的csv文件
file_paths = glob.glob('./dags/data/data_*.csv')
# 提取纯文件名，去掉路径前缀
files = [os.path.basename(path) for path in file_paths]

如果要匹配其他规则，修改通配符规则即可：

• 匹配所有csv文件：*.csv
• 匹配以2开头的csv文件：2*.csv
• 匹配包含test字符的csv文件：*test*.csv

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