如何在Airflow Bash Operator中自动引入Airflow配置值

1. 能否添加Airflow配置值作为环境变量？ #

完全可以。Airflow会将配置文件里的参数自动转换为AIRFLOW__<配置段>__<配置项>格式的环境变量，比如配置中[core]段的executor对应环境变量AIRFLOW__CORE__EXECUTOR。

你可以直接在Bash Operator的env参数中指定这些变量，或者直接在bash命令里引用它们。举个例子：

from airflow.operators.bash import BashOperator

# 手动指定配置变量
bash_task = BashOperator(
    task_id="list_dags",
    bash_command="airflow dags list",
    env={
        "AIRFLOW__CORE__EXECUTOR": "CeleryExecutor",
        "AIRFLOW__DATABASE__SQL_ALCHEMY_CONN": "postgresql+psycopg2://user:pass@db:5432/airflow"
    }
)

如果要复用当前Airflow实例的现有配置值，也可以通过os.environ获取后传入：

import os
from airflow.operators.bash import BashOperator

bash_task = BashOperator(
    task_id="show_executor",
    bash_command="echo 当前Airflow执行器: $AIRFLOW__CORE__EXECUTOR",
    env={"AIRFLOW__CORE__EXECUTOR": os.environ.get("AIRFLOW__CORE__EXECUTOR")}
)

2. 能否无需逐一列出传递所有环境变量？ #

当然可以。你可以直接把当前进程的环境变量副本传给Bash Operator的env参数，这样就不用逐个列举：

import os
from airflow.operators.bash import BashOperator

bash_task = BashOperator(
    task_id="run_airflow_test",
    bash_command="airflow tasks test my_dag my_task 2023-01-01",
    env=os.environ.copy()
)

不过这么做会把worker进程的所有环境变量都传过去，可能包含一些无关变量。如果只想传递Airflow相关的变量，可以过滤出以AIRFLOW__开头的变量：

import os
from airflow.operators.bash import BashOperator

# 只保留Airflow相关环境变量
airflow_only_env = {k: v for k, v in os.environ.items() if k.startswith("AIRFLOW__")}
bash_task = BashOperator(
    task_id="show_airflow_info",
    bash_command="airflow info",
    env=airflow_only_env
)

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