创建每日DAG处理动态文件夹列表的技术问题咨询

核心思路是利用Airflow 2.3+引入的**动态任务映射（Dynamic Task Mapping）**特性，它能在DAG运行阶段根据实际数据生成任务，完美解决解析阶段无法获取当日文件夹列表的问题。同时借助Airflow内置的上下文变量固定启动日期，避免全局变量随时间变更的问题。

步骤1：固定DAG启动日期 #

放弃使用datetime.now()这类会随解析时间变化的全局变量，直接通过DAG Run的上下文获取固定的启动日期：

• 使用context['execution_date'].date()或context['data_interval_start'].date()，这两个值在DAG启动后就固定为当日日期，即使任务跨多日运行也不会改变。

步骤2：编写获取文件夹列表的上游任务 #

先创建一个PythonOperator任务，专门读取启动日期对应的文件夹列表文件，并将结果通过XCom传递给下游任务：

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta
import os

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'daily_folder_processor',
    default_args=default_args,
    schedule_interval='@daily',
    catchup=False,
    tags=['folder_processing'],
) as dag:

    def get_folder_list(**context):
        # 获取本次DAG Run的固定启动日期
        run_date = context['execution_date'].date()
        # 按日期构造文件夹列表文件路径（根据你的实际路径调整）
        list_file_path = f'/data/folder_lists/folders_{run_date}.txt'
        
        if not os.path.exists(list_file_path):
            raise ValueError(f"Folder list file missing for date: {run_date}")
        
        # 读取文件，每行一个文件夹路径
        with open(list_file_path, 'r') as f:
            folder_list = [line.strip() for line in f if line.strip()]
        
        return folder_list

    fetch_folders = PythonOperator(
        task_id='fetch_daily_folder_list',
        python_callable=get_folder_list,
        provide_context=True,
    )

步骤3：动态生成每个文件夹的处理任务 #

通过partial()和expand()方法，基于上游任务返回的文件夹列表，动态生成对应的PythonOperator和BashOperator任务：

# 动态生成Python处理任务
    process_with_python = PythonOperator.partial(
        task_id='process_folder_python',
        python_callable=lambda folder: print(f"Processing folder {folder} with Python logic..."),
        provide_context=True,
    ).expand(op_kwargs=[{'folder': folder} for folder in '{{ ti.xcom_pull(task_ids="fetch_daily_folder_list") }}'])

    # 动态生成Bash处理任务
    process_with_bash = BashOperator.partial(
        task_id='process_folder_bash',
        bash_command='echo "Processing folder {{ params.folder }} with Bash script..."',
    ).expand(params=[{'folder': folder} for folder in '{{ ti.xcom_pull(task_ids="fetch_daily_folder_list") }}'])

    # 设置任务依赖：先获取文件夹列表，再并行处理所有文件夹
    fetch_folders >> [process_with_python, process_with_bash]

关键说明 #

1. 为什么之前的循环无效？
   你之前尝试的循环是在DAG解析阶段执行的（Airflow调度器每隔一段时间会重新解析DAG文件），此时还没有具体的DAG Run实例，无法获取对应日期的文件夹列表。而动态任务映射是在DAG Run运行阶段执行的，完全基于本次Run的上下文数据生成任务。

2. 任务自动命名
   Airflow会为每个动态生成的任务自动添加后缀（如process_folder_python__0、process_folder_python__1），确保任务ID唯一。

3. 大列表场景优化
   如果文件夹数量极大（上千个），需要注意：

   • 调整Airflow配置中的max_xcom_size，避免XCom存储超限；
   • 也可以将文件夹列表存入外部存储（如数据库、Redis），下游任务直接从外部读取，而非依赖XCom。

4. 并行度控制
   根据集群资源情况，调整DAG的max_active_tasks参数，避免同时运行过多任务导致资源耗尽。

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