Airflow问题：如何避免DAG在导入阶段立即执行任务

你的核心问题是DAG导入阶段执行了get_items()函数，导致API调用在Airflow扫描DAG文件时就触发，一旦API失败会直接导致DAG标记为异常。要实现触发后才动态生成任务，推荐使用Airflow的动态任务映射（Dynamic Task Mapping）（Airflow 2.2及以上版本支持），这是官方推荐的动态任务生成方案。

问题根源 #

你当前代码中for item in get_items()是在DAG解析阶段执行的，Airflow会定期扫描/dags目录加载DAG文件，这段代码会被立即执行，导致API请求提前触发。

修改后的代码 #

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta
import requests

# Default args for the DAG
default_args = {
    'owner': 'me',
    'start_date': datetime(2025, 1, 1),
    'depends_on_past': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# Create a DAG instance
dag = DAG(
    'my_dag_id',
    default_args=default_args,
    schedule=None,
    catchup=False,  # 禁止回溯执行
)

def get_items():
    """
    Makes a HTTP request to an API,
    retrieves a list of items from the response,
    and returns the list
    """
    try:
        response = requests.get('https://api.example.com/items')
        response.raise_for_status()  # 主动触发HTTP错误异常
        items = response.json()['items']
        return items
    except requests.exceptions.RequestException as e:
        # 捕获API请求异常，添加日志或告警逻辑
        raise ValueError(f"API调用失败: {str(e)}") from e

def process_item(item):
    """
    Processes a single item
    """
    print(f'Processing item {item}')

# 创建获取条目的任务，结果自动存入XCom
get_items_task = PythonOperator(
    task_id='get_items',
    python_callable=get_items,
    dag=dag,
)

# 动态生成处理任务：通过expand接收上游任务的输出，自动为每个item创建子任务
process_items_task = PythonOperator(
    task_id='process_item',
    python_callable=process_item,
    op_args=[lambda item: item],
    dag=dag,
).expand(op_args=get_items_task.output)

# 设置依赖关系
get_items_task >> process_items_task

关键说明 #

1. 避免导入阶段执行API调用：get_items()仅在get_items_task运行时执行（即DAG触发后），DAG导入阶段不会调用该函数。
2. 动态任务映射：通过.expand(op_args=get_items_task.output)，Airflow会在get_items_task执行完成后，根据返回的items列表自动生成对应数量的子任务，每个子任务处理一个item。
3. 异常处理：在get_items()中添加了异常捕获，确保API调用失败时仅触发任务失败，不会导致DAG导入异常，同时保留错误信息便于排查。
4. XCom存储：PythonOperator默认会将函数返回值存入XCom，get_items_task.output就是指向该XCom的引用，用于动态映射。

低版本Airflow兼容方案（2.2以下） #

如果你的Airflow版本低于2.2，无法使用动态任务映射，可以通过以下方式实现：

• 使用PythonOperator结合TaskGroup，在任务执行时动态创建子任务（需要调用Airflow内部API）
• 将条目列表存储到外部数据库，后续任务从数据库读取并处理（避免依赖XCom的大小限制）

不过更推荐升级到Airflow 2.2+版本，动态任务映射是官方原生支持的更简洁方案。

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