如何在单个Airflow DAG中切换两个不同的schema？

方案1：DAG运行时传参（推荐，和手动执行bash脚本逻辑完全匹配） #

直接给DAG配置可选择的参数项，触发时手动选择目标schema即可，支持手动触发和定时触发场景：

• DAG代码示例：

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

with DAG(
    dag_id="your_pre_schema_dag",
    start_date=datetime(2024, 1, 1),
    schedule=None, # 按需修改为对应定时cron表达式
    catchup=False,
    # 配置schema可选参数
    params={
        "target_schema": {
            "type": "string",
            "enum": ["mstr_new", "mstr_pre"],
            "default": "mstr_pre",
            "description": "选择PRE环境要执行的目标Schema"
        }
    }
) as dag:
    exec_sh_task = BashOperator(
        task_id="execute_script",
        bash_command="./script.sh {{ params.target_schema }}"
    )

• 触发DAG时，Airflow界面会自动生成schema下拉选择框，选中对应值后运行，参数会自动代入到bash命令中。
  如果使用Airflow 1.x版本不支持原生params功能，触发时可在运行配置的conf项中填入{"target_schema": "mstr_new"}，命令中通过{{ dag_run.conf.get('target_schema', 'mstr_pre') }}读取参数即可

方案2：Airflow全局变量（适合不频繁切换schema的场景） #

如果不需要每次运行都调整schema，只是偶尔切换，可以通过Airflow全局变量配置：

1. 进入Airflow界面的Admin -> Variables页面，新增变量：key设为pre_target_schema，value设为需要使用的schema（mstr_new/mstr_pre）
2. 任务中直接读取变量即可，代码示例：

from airflow.models import Variable
from airflow.operators.bash import BashOperator

exec_sh_task = BashOperator(
    task_id="execute_script",
    bash_command=f"./script.sh {Variable.get('pre_target_schema')}"
)

需要切换schema时直接修改全局变量的值即可，无需调整DAG代码。

方案3：单定义多DAG实体（适合两个schema有独立触发规则的场景） #

如果两个schema的执行时间、依赖逻辑不同，可以用同一份代码生成两个独立DAG，各自对应不同的schema：

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

schemas = ["mstr_new", "mstr_pre"]

for schema in schemas:
    with DAG(
        dag_id=f"pre_schema_{schema}_dag",
        start_date=datetime(2024, 1, 1),
        # 可根据schema不同设置独立的定时规则
        schedule="0 1 * * *" if schema == "mstr_pre" else "0 3 * * *",
        catchup=False
    ) as dag:
        exec_sh_task = BashOperator(
            task_id=f"execute_{schema}_script",
            bash_command=f"./script.sh {schema}"
        )
    # 注册DAG到全局作用域
    globals()[f"pre_dag_{schema}"] = dag

该方式会在Airflow界面生成两个独立的DAG，各自执行对应schema的任务，互不干扰。

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