如何通过编程判断Airflow DAG处于解析状态还是触发状态？

我完全懂你的困扰——Airflow的调度器每隔30秒就会扫描解析一遍DAG文件，要是把环境准备的代码直接写在DAG文件的顶级位置，那每次解析都会跑一遍，完全是无效消耗。要区分解析态和运行态，这里有几个实用的方法：

方法1：检查Airflow运行时环境变量 #

当DAG被触发执行时，Airflow会自动注入一批以AIRFLOW_CTX_开头的环境变量（比如AIRFLOW_CTX_DAG_RUN_ID、AIRFLOW_CTX_TASK_ID），而解析阶段这些变量是不存在的。你可以通过判断这些变量来控制逻辑执行：

import os

# 仅在DAG运行时执行环境准备
if "AIRFLOW_CTX_DAG_RUN_ID" in os.environ:
    print("进入DAG运行时，开始执行环境准备...")
    # 调用你的第三方Python库完成准备工作
    # your_external_library.setup_environment()
else:
    # 解析阶段，直接跳过
    pass

方法2：通过Airflow上下文对象判断 #

Airflow在运行时（任务执行或DAG启动阶段）会生成上下文对象，而解析阶段调用get_current_context()会抛出RuntimeError。我们可以利用这个特性来区分状态：

from airflow.utils.context import get_current_context

try:
    # 尝试获取运行时上下文
    context = get_current_context()
    # 能拿到dag_run对象说明处于运行态
    if context.get("dag_run") is not None:
        your_environment_prep_function()
except RuntimeError:
    # 解析阶段捕获错误，直接跳过
    pass

方法3：使用Airflow原生DAG钩子（最优雅） #

Airflow的DAG定义支持on_dag_run_start钩子，这个钩子只会在DAG被触发启动时执行，解析阶段完全不会触发，完美匹配你的需求：

from airflow import DAG
from datetime import datetime

# 定义环境准备函数
def prepare_runtime_environment(context):
    dag_run_id = context["dag_run"].run_id
    print(f"为DAG Run {dag_run_id} 初始化环境...")
    # 调用第三方库完成准备
    # your_external_library.setup()

# 定义DAG时绑定钩子
with DAG(
    dag_id="your_target_dag",
    start_date=datetime(2024, 1, 1),
    schedule_interval="@daily",
    on_dag_run_start=prepare_runtime_environment  # 关键配置
) as dag:
    # 这里编写你的任务逻辑
    # ...

小提醒：如果你的环境准备需要生成后续任务依赖的资源（比如配置文件、临时数据），要确保资源存储在Airflow Worker能访问到的路径（比如共享存储、分布式文件系统），避免任务出现资源找不到的问题。

备注：内容来源于stack exchange，提问作者alex