如何在Airflow中遍历数据并为每条数据启动独立DAG实例

核心思路 #

直接通过Airflow内部ORM（DagRun模型）在采集DAG中异步触发目标DAG实例，而非使用DagRunOperator（后者会作为子任务占用worker，易引发资源耗尽问题）。采集DAG仅负责触发，不等待目标DAG的执行结果，完全解耦两个DAG的资源占用。

方案优势 #

• 避免子任务带来的worker资源竞争与死锁风险
• 触发逻辑异步执行，采集DAG执行效率不受目标DAG运行时长影响
• 可通过速率限制、队列隔离等手段轻松扩展到上千条数据的场景

代码实现 #

1. 目标DAG（单条目处理逻辑）

这个DAG负责处理单个数据条目，仅接受手动/外部触发：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def process_single_item(**context):
    # 从DagRun的配置中获取当前条目数据
    item = context["dag_run"].conf.get("item")
    # 替换为你的实际业务逻辑：存库、计算、调用第三方接口等
    print(f"Processing item ID {item['id']}: {item['name']}")

with DAG(
    dag_id="process_single_item",
    schedule_interval=None,  # 禁止自动调度，仅接受外部触发
    start_date=datetime(2023, 1, 1),
    catchup=False,
    default_args={"queue": "item_processing_queue"},  # 配置专属队列，隔离资源
    tags=["item_processing"]
) as dag:
    process_task = PythonOperator(
        task_id="execute_item_processing",
        python_callable=process_single_item,
        provide_context=True
    )

2. 采集DAG（数据拉取+批量触发）

这个DAG负责从API拉取数据，并遍历触发目标DAG的实例：

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.models import DagRun
from airflow.utils.state import DagRunState
from datetime import datetime
import time
from typing import List, Dict

# 模拟API数据采集，替换为你的实际API调用逻辑
def fetch_api_data() -> List[Dict]:
    return [
        {"id": 1, "name": "product_A", "price": 99.9},
        {"id": 2, "name": "product_B", "price": 199.9},
        # 支持任意数量的数据条目
    ]

def trigger_item_processing_dags(**context):
    items = fetch_api_data()
    target_dag_id = "process_single_item"
    trigger_delay = 0.3  # 每0.3秒触发一个，避免短时间内压垮Airflow调度系统

    for item in items:
        try:
            # 生成唯一的run_id，避免重复触发
            run_id = f"manual_trigger_{datetime.now().strftime('%Y%m%d%H%M%S')}_{item['id']}"
            # 异步创建DagRun实例，直接加入调度队列
            DagRun.create(
                dag_id=target_dag_id,
                run_id=run_id,
                conf={"item": item},  # 传递当前条目数据
                state=DagRunState.QUEUED
            )
            print(f"Successfully triggered DagRun for item {item['id']}")
        except Exception as e:
            # 捕获触发异常，不中断整个采集流程
            print(f"Failed to trigger item {item['id']}: {str(e)}")
        # 速率限制，控制触发频率
        time.sleep(trigger_delay)

with DAG(
    dag_id="collect_and_trigger_items",
    schedule_interval="@daily",  # 按业务需求配置调度周期
    start_date=datetime(2023, 1, 1),
    catchup=False,
    default_args={"queue": "collection_queue"},  # 采集DAG使用独立队列
    tags=["data_collection"]
) as dag:
    trigger_task = PythonOperator(
        task_id="batch_trigger_item_dags",
        python_callable=trigger_item_processing_dags,
        provide_context=True
    )

扩展性优化建议 #

1. 资源隔离：给采集DAG和目标DAG配置不同的worker队列，避免互相抢占资源。在Airflow的worker配置中指定队列监听规则。
2. 动态速率控制：如果数据量波动大，可以根据当前Airflow的队列长度动态调整trigger_delay，比如队列任务多就延长延迟，队列空闲就缩短延迟。
3. 批量分块：对于上万条数据，可将数据分成若干块，每块触发后短暂休眠，避免单次任务运行时间过长。
4. 幂等性保障：在目标DAG的处理逻辑中加入幂等校验（比如通过item['id']判断是否已处理），防止重复触发导致的重复操作。
5. 监控告警：采集DAG中可加入触发成功/失败的统计，通过Airflow的日志或第三方监控系统发送告警。

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