如何在Apache Airflow中循环处理多个文件？技术实现问询

安全阿岳

2026-5-13

解决Airflow多文件逐个检查并触发DAG的问题

我看你现在的实现只能处理第一个匹配到的文件，没法对多个文件逐个触发目标DAG，下面给你调整方案，让你能实现遍历所有目标文件，每个存在的文件都带参数触发一次目标DAG：

首先分析现有代码的局限

你的execute_check_if_file_exists_task函数在找到第一个匹配文件后就直接return了，导致后续文件不会被检查；同时TriggerDagRunOperator本身只能触发一次DAG运行，没法批量处理多个文件的触发需求。

修改后的完整实现

1. 调整文件检查逻辑：收集所有匹配的文件信息

先把原来的分支任务改成收集所有符合条件的文件，而不是找到第一个就返回：

import os
import re
import logging
import time
from airflow.models import DagRun, DagModel
from airflow.operators.python import PythonOperator, BranchPythonOperator

def execute_check_if_file_exists_task(*args, **kwargs):
    input_file_list = ["a","b"]
    matched_files = []
    for item in input_file_list:
        full_path = json_data[item]['input_folder_path']
        # 先检查目录是否存在，避免报错
        if not os.path.exists(full_path):
            logging.warning(f"目录 {full_path} 不存在，跳过检查")
            continue
        # 遍历目录下的文件
        for file_name_in_dir in os.listdir(full_path):
            if re.match(file_name, file_name_in_dir):
                # 收集所有匹配的文件信息
                matched_files.append({
                    'file_type': item,
                    'file_name': file_name_in_dir,
                    'file_path': os.path.join(full_path, file_name_in_dir)
                })
    # 将所有匹配文件推送到XCom，供后续任务使用
    kwargs['ti'].xcom_push(key='matched_files', value=matched_files)
    # 分支判断：有文件就去触发DAG，否则进入未找到分支
    return "trigger_dag_run_task" if matched_files else "file_not_found_task"

2. 替换TriggerDagRunOperator：循环触发多个DAG实例

改用PythonOperator，通过Airflow内部API逐个触发目标DAG：

def trigger_multiple_dags(context):
    ti = context['task_instance']
    # 从XCom获取之前收集的所有匹配文件
    matched_files = ti.xcom_pull(task_ids='check_if_file_exists_task', key='matched_files')
    
    for file_info in matched_files:
        # 生成唯一的run_id，避免重复触发冲突
        run_id = f"triggered_{file_info['file_type']}_{file_info['file_name']}_{int(time.time())}"
        # 构造传递给目标DAG的参数
        dag_conf = {
            'file_type': file_info['file_type'],
            'file_name': file_info['file_name'],
            'file_path': file_info['file_path']
        }
        # 触发目标DAG
        DagRun.create(
            dag_id="trigger_dag",
            run_id=run_id,
            conf=dag_conf,
            external_trigger=True
        )
        logging.info(f"已触发DAG实例 {run_id}，对应文件：{file_info['file_path']}")

3. 重新定义Operator并关联任务

# 文件未找到的任务（保留原逻辑）
def execute_file_not_found_task(*args, **kwargs):
    logging.info("未找到匹配的文件路径。")

file_not_found_task = PythonOperator(
    task_id='file_not_found_task',
    retries=3,
    provide_context=True,
    dag=dag,
    python_callable=execute_file_not_found_task,
)

# 分支检查任务（使用修改后的逻辑）
check_if_file_exists_task = BranchPythonOperator(
    task_id='check_if_file_exists_task',
    retries=3,
    provide_context=True,
    dag=dag,
    python_callable=execute_check_if_file_exists_task,
)

# 多文件触发任务（替换原TriggerDagRunOperator）
trigger_dag_run_task = PythonOperator(
    task_id='trigger_dag_run_task',
    provide_context=True,
    dag=dag,
    python_callable=trigger_multiple_dags,
)

# 设置任务依赖
check_if_file_exists_task.set_downstream([trigger_dag_run_task, file_not_found_task])

关键修改点说明

收集所有匹配文件：不再提前返回，而是把所有符合条件的文件信息存入XCom，确保每个文件都能被处理
使用DagRun API触发：绕过TriggerDagRunOperator单次触发的限制，通过DagRun.create()循环创建多个DAG实例，每个实例携带对应文件的参数
唯一run_id：用文件类型、文件名+时间戳生成唯一标识，避免重复触发导致的冲突

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