Airflow Task Group条件逻辑异常：ENV为PROD/STAGE时仍始终进入else分支

我完全懂你这种摸不着头脑的感觉——明明在UI和CLI里都确认了ENV变量是PROD/STAGE，可Task Group的条件分支偏偏每次都走到else里，连DAG图都显示着不该有的任务依赖，这确实让人头大。咱们一步步拆解问题，找到根因和解决办法：

一、先排查最核心的问题：DAG解析时的ENV值到底是什么？ #

你提到已经打印过ENV值是PROD，但这里有个容易忽略的关键点：你手动查询的是当前元数据库里的变量值，但Airflow调度器在解析DAG文件时，获取的可能是另一个值（比如旧缓存、解析时元数据库连接异常、或者上下文不一致）。

你可以在DAG文件开头加上日志，把解析时的ENV值直接输出到调度器日志里，这样就能100%确认调度器看到的是什么值：

import logging
from airflow.models import Variable

# 配置日志
logger = logging.getLogger(__name__)

try:
    env = Variable.get("ENV")
    parsed_env = env.strip().upper()
    logger.info(f"=== 【DAG解析阶段】获取到的ENV变量值: {parsed_env} ===")
except Exception as e:
    logger.error(f"=== 【DAG解析阶段】获取ENV变量失败: {str(e)} ===")
    parsed_env = "DEV"  # 临时设置默认值，避免DAG解析失败

然后去Airflow调度器的日志里找这条打印信息，如果这里的parsed_env不是PROD/STAGE，那就能解释为什么else分支会被执行了——调度器根本没拿到你设置的变量值。

可能的原因包括：

• 调度器进程的元数据库连接临时异常，导致Variable.get返回空值或默认值
• Airflow的DAG解析缓存没更新（可以尝试重启调度器时加--reset-dag-run参数，或者删除DAG的本地缓存文件）
• 你设置的是DAG级别的变量而非全局变量（检查变量的dag_id字段是否为空）

二、你的Task Group任务定义逻辑有冗余 #

即使解决了ENV值的问题，当前的代码还有个小问题：不管是PROD还是其他环境，census_task和table_task都会被创建（因为创建任务的代码在if/else分支外面）。也就是说，即使是PROD环境，census_task还是会出现在Task Group里，只是没有上游依赖——这其实不符合你“只运行table_task”的需求。

正确的做法是把任务创建逻辑也放进条件分支里，让PROD环境下完全不创建census_task：

@task_group()
def human_capital_tasks():
    if parsed_env in ["PROD", "STAGE"]:
        # PROD/STAGE环境：只创建并保留table_task
        table_task = TriggerDagRunOperator(
            task_id="supervisor_table",
            trigger_dag_id="supervisor_table",
            wait_for_completion=True,
            trigger_rule=TriggerRule.ALL_DONE,
        )
    else:
        # 其他环境：创建两个任务并设置依赖
        census_task = TriggerDagRunOperator(
            task_id="term_census",
            trigger_dag_id="term_census",
            wait_for_completion=True,
            trigger_rule=TriggerRule.ALL_DONE,
        )
        table_task = TriggerDagRunOperator(
            task_id="supervisor_table",
            trigger_dag_id="supervisor_table",
            wait_for_completion=True,
            trigger_rule=TriggerRule.ALL_DONE,
        )
        census_task >> table_task

这样PROD环境下的Task Group里只会有table_task，不会出现多余的census_task。

三、更可靠的环境配置方案：用系统环境变量代替Airflow Variables #

Airflow Variables存储在元数据库里，DAG解析时需要访问元数据库，这就可能出现连接延迟、缓存、权限等问题，不太适合存储环境标识这种全局配置。

更推荐的方式是用系统环境变量来设置ENV：

1. 在调度器、webserver、worker的启动环境里设置ENV变量，比如Docker Compose里：

   services:
     airflow-scheduler:
       environment:
         - ENV=PROD
     airflow-webserver:
       environment:
         - ENV=PROD
     airflow-worker:
       environment:
         - ENV=PROD
   

2. 在DAG文件里直接读取系统环境变量：

   import os
   
   parsed_env = os.getenv("ENV", "DEV").strip().upper()
   

这种方式不需要依赖元数据库，DAG解析时直接读取进程环境变量，速度更快，也不会出现变量缓存或连接问题，更适合环境级别的配置。

总结解决步骤 #

1. 先通过日志确认调度器解析DAG时的ENV值，确保和你设置的一致
2. 调整Task Group的任务创建逻辑，把任务定义放进条件分支里
3. （可选）切换为系统环境变量来管理ENV配置，避免元数据库依赖问题

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