Kubernetes部署Airflow：Operator无法读取Airflow变量问题

PythonVirtualenvOperator会创建一个完全独立的Python虚拟环境，该环境默认不会加载Airflow的元数据配置，因此直接在虚拟环境内调用Variable.get()无法读取Web UI中配置的变量。以下是几种可行的解决方法：

方法一：通过op_kwargs直接传递变量 #

这是最简洁高效的方案，在Operator外部先获取变量，再通过op_kwargs参数传给虚拟环境内的函数：

from airflow.models import Variable
from airflow.operators.python import PythonVirtualenvOperator

def process_data(my_custom_var):
    # 直接使用传入的变量值
    print(f"获取到的变量值：{my_custom_var}")

virtualenv_task = PythonVirtualenvOperator(
    task_id="run_in_virtualenv",
    python_callable=process_data,
    op_kwargs={
        "my_custom_var": Variable.get("MY_SUPER_DOOPER_VAR")
    },
    # 若虚拟环境需要额外依赖，在这里添加
    requirements=["pandas==2.1.0"],
    dag=dag
)

方法二：在虚拟环境中加载Airflow配置获取变量 #

如果必须在虚拟环境内部调用Variable.get()，需要将Airflow的核心配置（如元数据库连接、加密密钥）传递到虚拟环境，并安装必要依赖：

from airflow.operators.python import PythonVirtualenvOperator
import os

def fetch_variable():
    from airflow.models import Variable
    var_value = Variable.get("MY_SUPER_DOOPER_VAR")
    print(f"获取到的变量值：{var_value}")

virtualenv_task = PythonVirtualenvOperator(
    task_id="fetch_var_in_virtualenv",
    python_callable=fetch_variable,
    # 安装Airflow及对应元数据库驱动（以PostgreSQL为例）
    requirements=["apache-airflow==2.8.0", "psycopg2-binary==2.9.9"],
    # 传递Airflow核心配置环境变量
    env_vars={
        "AIRFLOW__CORE__SQL_ALCHEMY_CONN": os.environ.get("AIRFLOW__CORE__SQL_ALCHEMY_CONN"),
        "AIRFLOW__CORE__FERNET_KEY": os.environ.get("AIRFLOW__CORE__FERNET_KEY")
    },
    dag=dag
)

注意：虚拟环境内安装的Airflow版本必须与集群部署的版本一致，避免兼容性问题。

方法三：使用Kubernetes Secret或Airflow Connection替代（敏感信息优先） #

如果变量包含敏感数据，建议将其存储在Kubernetes Secret或Airflow Connection中，通过环境变量传递到虚拟环境：

from airflow.operators.python import PythonVirtualenvOperator

def use_secret_var():
    import os
    secret_value = os.getenv("MY_SUPER_SECRET_VAR")
    print(f"获取到的敏感变量：{secret_value}")

virtualenv_task = PythonVirtualenvOperator(
    task_id="use_secret_in_virtualenv",
    python_callable=use_secret_var,
    env_vars={
        # 从Airflow Connection中读取
        "MY_SUPER_SECRET_VAR": "{{ conn.my_prod_conn.password }}",
        # 或从Kubernetes Secret中读取（需确保Worker Pod有权限访问）
        # "MY_SUPER_SECRET_VAR": "{{ var.value.my_k8s_secret }}"
    },
    dag=dag
)

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