Airflow DAG如何根据dev/prod环境设置不同调度周期？

完全可行，你可以通过以下几种常见方式实现：

• 利用环境变量区分环境
  在Airflow运行环境中设置环境变量（比如AIRFLOW_ENV），值为dev或prod。然后在DAG定义中读取该变量，动态设置调度时间：

  import os
  from airflow import DAG
  from datetime import datetime
  
  env = os.getenv("AIRFLOW_ENV", "dev")  # 默认设为开发环境
  
  schedule_interval = "*/10 * * * *" if env == "dev" else "@hourly"
  
  with DAG(
      dag_id="env_based_schedule_dag",
      start_date=datetime(2024, 1, 1),
      schedule_interval=schedule_interval,
      catchup=False
  ) as dag:
      # 这里定义你的具体任务
      pass
  

• 使用Airflow Variables存储环境配置
  在Airflow UI的「Admin > Variables」中添加名为env的变量，值设为dev或prod。之后在DAG中读取该变量：

  from airflow.models import Variable
  from airflow import DAG
  from datetime import datetime
  
  env = Variable.get("env", default_var="dev")
  schedule_interval = "*/10 * * * *" if env == "dev" else "@hourly"
  
  with DAG(
      dag_id="var_based_schedule_dag",
      start_date=datetime(2024, 1, 1),
      schedule_interval=schedule_interval,
      catchup=False
  ) as dag:
      # 任务定义
      pass
  

• 通过Airflow配置文件区分
  在airflow.cfg中添加自定义配置项：

  [custom]
  environment = dev
  

  然后在DAG中读取配置值：

  from airflow.configuration import conf
  from airflow import DAG
  from datetime import datetime
  
  env = conf.get("custom", "environment", fallback="dev")
  schedule_interval = "*/10 * * * *" if env == "dev" else "@hourly"
  
  with DAG(
      dag_id="conf_based_schedule_dag",
      start_date=datetime(2024, 1, 1),
      schedule_interval=schedule_interval,
      catchup=False
  ) as dag:
      # 任务定义
      pass
  

优先推荐环境变量的方式，这种方式在Docker、K8s等容器化部署场景中更便于管理和切换环境。

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