未记录事实核验 · 状态：维护中

记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。

变量连接与配置管理

---

为什么需要集中化的配置管理

不好的做法（硬编码在 DAG 文件里）：

# 危险！密码硬写在代码里，会被 git 记录，有安全风险
conn = mysql.connect(
    host="192.168.1.100",
    user="etl_user",
    password="my_secret_password_123",  # 绝对不要这样做
    database="business_db"
)

好的做法（通过 Airflow 的配置管理）：

# 安全：Connection 信息存在 MetaDB 中（可加密），代码里只有连接 ID
from airflow.hooks.base import BaseHook
conn_info = BaseHook.get_connection("mysql_prod")

---

Variable：全局配置参数

Variable 适合存储非敏感的配置信息，可以在 Web UI 中直接修改，不需要改代码。

创建和管理 Variable

方式 1：Web UI 创建

Admin → Variables → 点击 "+" 按钮，填写 Key 和 Value

方式 2：命令行创建

# 创建简单字符串变量
airflow variables set target_env "production"

# 创建 JSON 变量
airflow variables set report_config '{"to_email": ["alice@company.com"], "cc": ["bob@company.com"]}'

# 从文件导入变量（批量导入）
airflow variables import variables.json

# 查看变量
airflow variables get target_env
airflow variables list

variables.json 示例：

{
    "target_env": "production",
    "dw_database": "bigdata_dw",
    "ods_table_prefix": "ods_",
    "dws_table_prefix": "dws_",
    "alert_email_list": ["data-team@company.com", "manager@company.com"],
    "data_retention_days": 90,
    "spark_queue": "etl_high_priority"
}

在 DAG 代码中读取 Variable

from airflow.models import Variable

# 读取字符串变量
env = Variable.get("target_env")
print(f"运行环境：{env}")

# 读取 JSON 变量（自动反序列化）
email_list = Variable.get("alert_email_list", deserialize_json=True)
# email_list = ["data-team@company.com", "manager@company.com"]

# 读取带默认值（变量不存在时不报错）
retention_days = Variable.get("data_retention_days", default_var=30)

# 读取 JSON 的某个字段
config = Variable.get("report_config", deserialize_json=True)
to_email = config.get("to_email", [])

在 Operator 参数的模板中使用 Variable

# 在 bash_command 的 Jinja 模板中读取 Variable
extract_task = BashOperator(
    task_id="extract_data",
    bash_command="""
        python extract.py \
            --env {{ var.value.target_env }} \
            --database {{ var.value.dw_database }} \
            --date {{ ds }}
    """,
)

# 读取 JSON 变量的字段
query_task = SQLExecuteQueryOperator(
    task_id="query_data",
    conn_id="mysql_prod",
    sql="SELECT * FROM {{ var.value.ods_table_prefix }}orders WHERE dt = '{{ ds }}'",
)

# 不推荐：在文件顶层读取（解析时会触发变量查找，是否访问 MetaDB 取决于后端与缓存配置）
TARGET_TABLE = Variable.get("target_table")  # 不要这样做

# 正确：在函数内部读取（只在 Task 执行时才查询）
def run_etl(**context):
    target_table = Variable.get("target_table")  # 在这里读取
    # ...

---

Connection：数据源连接管理

Connection 用于存储数据库、API、存储系统的连接信息，包含主机、端口、用户名、密码等。

创建 Connection

方式 1：Web UI 创建

Admin → Connections → 点击 "+"

字段	MySQL 示例	PostgreSQL 示例
Conn Id	mysql_prod	postgres_dw
Conn Type	MySQL	Postgres
Host	192.168.1.100	192.168.1.101
Schema	business_db	data_warehouse
Login	etl_user	dw_reader
Password	********	********
Port	3306	5432

方式 2：环境变量创建（推荐生产环境）

# 格式：AIRFLOW_CONN_{CONN_ID}=conn_type://user:password@host:port/schema
# 环境变量名通常将 conn_id 转为大写，并保留下划线

export AIRFLOW_CONN_MYSQL_PROD="mysql://etl_user:my_password@192.168.1.100:3306/business_db"
export AIRFLOW_CONN_POSTGRES_DW="postgresql://dw_reader:password@192.168.1.101:5432/data_warehouse"
export AIRFLOW_CONN_HIVE_DEFAULT="hive://user@hiveserver2:10000/default"

# JSON 格式（更灵活，支持 extra 参数）
export AIRFLOW_CONN_SPARK_DEFAULT='{
    "conn_type": "spark",
    "host": "spark://spark-master:7077",
    "extra": "{\"deploy-mode\": \"cluster\", \"spark-home\": \"/opt/spark\"}"
}'

方式 3：命令行创建

airflow connections add mysql_prod \
    --conn-type mysql \
    --conn-host 192.168.1.100 \
    --conn-login etl_user \
    --conn-password "my_password" \
    --conn-schema business_db \
    --conn-port 3306

# 查看连接列表
airflow connections list

# 查看某个连接；输出可能涉及敏感信息，勿直接写入共享日志
airflow connections get mysql_prod

在 DAG 中使用 Connection

方式 1：在 Operator 中直接引用 conn_id

# 大多数 Provider Operator 直接接受 conn_id
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from airflow.providers.apache.hive.operators.hive import HiveOperator

sql_task = SQLExecuteQueryOperator(
    task_id="query_mysql",
    conn_id="mysql_prod",  # 直接引用 Connection ID
    sql="SELECT COUNT(*) FROM orders WHERE dt = '{{ ds }}'",
)

hive_task = HiveOperator(
    task_id="run_hql",
    hive_cli_conn_id="hive_default",
    hql="SELECT ...",
)

方式 2：通过 Hook 获取连接对象

from airflow.hooks.base import BaseHook
from sqlalchemy import create_engine

def run_custom_query(**context):
    # 获取 Connection 对象
    conn = BaseHook.get_connection("mysql_prod")

    # 用 Hook 构建引擎，避免自行拼接含密码的 URL
    from airflow.providers.mysql.hooks.mysql import MySqlHook
    from sqlalchemy import text
    engine = MySqlHook(mysql_conn_id="mysql_prod").get_sqlalchemy_engine()

    with engine.connect() as connection:
        result = connection.execute(text("SELECT COUNT(*) FROM orders"))
        count = result.scalar()
        print(f"订单数：{count}")

    return count

方式 3：使用 Hook（推荐，封装了连接细节）

from airflow.providers.mysql.hooks.mysql import MySqlHook
from airflow.providers.postgres.hooks.postgres import PostgresHook

def query_with_hook(**context):
    # MySqlHook 封装了连接管理
    mysql_hook = MySqlHook(mysql_conn_id="mysql_prod")

    # 直接执行查询，返回 pandas DataFrame
    df = mysql_hook.get_pandas_df(
        sql="SELECT * FROM orders WHERE dt = %(date)s",
        parameters={"date": context["ds"]}
    )
    print(f"查询到 {len(df)} 行数据")

    # 向 PostgreSQL 写入数据
    pg_hook = PostgresHook(postgres_conn_id="postgres_dw")
    pg_hook.insert_rows(
        table="ods_orders",
        rows=df.values.tolist(),
        target_fields=df.columns.tolist(),
        replace=False  # 示例仅演示插入；需要更新已有记录时须另行设计写入语义
    )

---

Secrets Backend：对接企业密钥管理

在企业环境中，通常有统一的密钥管理系统（如 HashiCorp Vault、AWS Secrets Manager），Airflow 支持从这些系统读取 Variable 和 Connection。

配置 HashiCorp Vault

# airflow.cfg
[secrets]
backend = airflow.providers.hashicorp.secrets.vault.VaultBackend
backend_kwargs = {
    "connections_path": "airflow/connections",
    "variables_path": "airflow/variables",
    "url": "http://vault:8200",
    "auth_type": "token",
    "token": "仅作示例的占位值，勿在配置文件中填写真实 token"
}

在 Vault 中存储 Connection：

# Vault 中的 KV 结构
vault kv put airflow/connections/mysql_prod \
    conn_type="mysql" \
    host="192.168.1.100" \
    login="etl_user" \
    password="my_secret" \
    schema="business_db" \
    port=3306

配置 AWS Secrets Manager

# airflow.cfg
[secrets]
backend = airflow.providers.amazon.aws.secrets.secrets_manager.SecretsManagerBackend
backend_kwargs = {
    "connections_prefix": "airflow/connections",
    "variables_prefix": "airflow/variables",
    "profile_name": "default"
}

Secrets 查找优先级

在未启用 worker 专用 Secrets Backend 的常见配置中，Variable 和 Connection 的查找顺序如下；支持并配置 worker 专用 Secrets Backend 的版本，在 worker 中还会优先查找该后端，须按实际版本与运行组件核对：

1. 配置的自定义 Secrets Backend（Vault、AWS SM 等）
2. 环境变量（AIRFLOW_CONN_xxx / AIRFLOW_VAR_xxx）
3. MetaDB（Web UI 中配置的）

---

环境变量：最轻量的配置方式

对于简单场景，直接用 OS 环境变量也是一种方式：

# 在 docker-compose.yaml 或系统环境中设置
export DB_HOST="192.168.1.100"
export DB_PASSWORD="my_password"

import os

def run_task(**context):
    # 从环境变量读取
    db_host = os.environ.get("DB_HOST", "localhost")
    db_password = os.environ.get("DB_PASSWORD")

    if not db_password:
        raise ValueError("DB_PASSWORD 环境变量未设置")

在 Docker Compose 中注入环境变量：

# docker-compose.override.yaml
services:
  airflow-worker:
    environment:
      - DB_HOST=192.168.1.100
      - DB_PASSWORD=${DB_PASSWORD}  # 从 .env 文件读取

---

配置管理最佳实践总结

配置类型	推荐方案	原因
非敏感业务配置	Airflow Variable	Web UI 可修改，不用改代码
数据库连接（含密码）	Connection + Secrets Backend	凭据不进代码；具体加密与访问控制取决于存储后端及其配置
环境区分（dev/prod）	环境变量或不同的 conn_id	避免测试和生产混用
API Key、Token	Secrets Backend	便于集中管理密钥；合规性仍取决于具体配置和组织要求
Spark 配置参数	Variable	调优时不需要改代码
邮件地址列表	Variable（JSON 格式）	灵活调整，不需要发版

---

小结

• Variable：非敏感配置的首选，Web UI 可管理；注意不要在 DAG 文件顶层读取（影响解析性能）
• Connection：数据源连接的标准方式；MetaDB 中密码的加密依赖正确配置 Fernet，其他后端的保护机制取决于其配置；生产环境可通过环境变量或 Secrets Backend 注入，并检查凭据暴露范围
• Secrets Backend：企业级密钥管理，对接 Vault/AWS SM；中大型团队的正确方案
• 永远不要把密码硬编码在 DAG 文件或 git 仓库中

下一篇：09-监控告警与故障处理 — Web UI 使用、任务重试、告警通知和故障排查

---

所属系列：00-Airflow学习路线图 | 上一篇：07-调度策略与时间管理 | 下一篇：09-监控告警与故障处理

#Airflow #Variable #Connection #Secrets #配置管理 #安全

---

---

掌握检查

• [ ] 能在 Web UI 中创建一个 MySQL Connection，并在 PythonOperator 中通过 BaseHook.get_connection() 读取它
• [ ] 能解释为什么不应该在 DAG 文件顶层调用 Variable.get()，而应该在 Task 函数内部调用
• [ ] 知道未启用 worker 专用 Secrets Backend 时的查找优先级：自定义 Secrets Backend > 环境变量 > MetaDB；启用 worker 专用后端时，能按实际版本核对 worker 中的查找顺序
• [ ] 能用环境变量覆盖 airflow.cfg 配置（AIRFLOW__SECTION__KEY 格式），举出至少 2 个实际例子
• [ ] 理解 Fernet Key 的作用——它用于加密 MetaDB 中受支持的敏感字段，如 Connection 密码，而不是加密 DAG 文件

---

实践练习

基础题：在 Airflow Web UI 中创建一个 Variable batch_size，值为 1000；然后在 DAG 的 PythonOperator 中读取这个变量，并打印出来。修改 Variable 值后不改代码重新触发 DAG，确认读取到了新值。

进阶题：如果你的公司要求数据库密码不能存在 Airflow MetaDB 中（避免被数据库管理员直接查到），应该怎么处理？列出至少两种方案，并说明各自的适用场景（提示：环境变量 vs Secrets Backend）。