基于Airflow环境变量实现差异化分支执行的可行性咨询

你提供的代码可以在dev、test、prod三个环境复用，能够实现Task_B仅在非dev环境运行的需求，以下是具体说明和注意事项：

核心逻辑有效性 #

你的代码通过Variable.get("Environment")获取环境标识，在DAG解析阶段判断环境是否为dev：

• 当环境为dev时，不会生成Start_dag_task >> Task_B >> End_dag_task这条依赖链，Task_B不会被调度执行；
• 当环境为test或prod时，会正常创建该依赖链，Task_B会按流程运行。

这种方式本质是利用Airflow的DAG解析机制，在不同环境下生成不同的任务依赖结构，实现分支控制。

关键注意事项 #

• 解析时机限制：Variable.get的取值是在DAG解析阶段完成的，如果中途修改Environment变量的值，需要触发DAG重新解析（比如重启scheduler或在Airflow UI手动刷新DAG）才能生效。
• 任务可见性差异：dev环境下，Task_B不会出现在DAG的任务依赖图中；非dev环境中，Task_B会正常显示并参与调度。
• 变量配置可靠性：需确保Airflow UI的Variables中已正确配置Environment变量，且取值严格为dev/test/prod，避免因变量缺失或取值错误导致逻辑异常。

可选优化方案 #

如果希望Task_B在dev环境中依然可见（仅标记为跳过），而非直接不生成依赖，可以改用任务内判断的方式：

from airflow.exceptions import AirflowSkipException

def task_b_func(**context):
    env = Variable.get("Environment")
    if env == "dev":
        raise AirflowSkipException("Dev环境下跳过Task_B")
    # 此处编写Task_B的实际业务逻辑

Task_B = PythonOperator(
    task_id="Task_B",
    python_callable=task_b_func,
    provide_context=True
)

# 统一配置依赖关系
Start_dag_task >> [Task_A, Task_B, Task_C] >> End_dag_task

这种方式能保留完整的DAG任务结构，便于查看全流程，同时实现dev环境下跳过Task_B的需求。

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