6.5 Apache Airflow

6.5 Apache Airflow：企业级工作流编排与调度平台

Apache Airflow 是一个成熟、可扩展的开源工作流编排系统，专为定义、调度、执行和监控复杂数据管道而设计。它以代码即配置（Code-as-Configuration）为核心理念，通过 Python 脚本声明式构建有向无环图（DAG），实现任务依赖的精确建模、执行状态的实时可视化以及故障的自动恢复。广泛应用于 ETL 流程、机器学习训练流水线、数据质量检查、定时报表生成等场景，已成为现代数据平台不可或缺的调度中枢。

核心架构与关键概念

Airflow 的能力源于其清晰分层的设计模型，核心组件协同完成端到端工作流管理：

关键演进提示：自 Airflow 2.0 起，SubDagOperator 已被标记为不推荐使用（Deprecated），官方强烈建议采用 TriggerDagRunOperator 或动态 DAG 生成替代，以规避性能瓶颈与状态管理复杂性。

快速实践：从零构建可运行 DAG

1. 环境初始化

# 创建隔离环境（推荐）
python -m venv airflow_env
source airflow_env/bin/activate  # Linux/macOS
# airflow_env\Scripts\activate  # Windows

# 安装 Airflow（指定版本以保障稳定性）
pip install "apache-airflow[postgres,redis]==2.8.1"

# 初始化数据库并启动 Webserver 与 Scheduler
airflow db upgrade
airflow users create \
    --username admin \
    --password admin \
    --firstname Admin \
    --lastname User \
    --role Admin \
    --email admin@example.com
airflow webserver &  # 后台启动 Web UI（默认 http://localhost:8080）
airflow scheduler   # 启动调度器（保持前台运行）
​

2. 定义生产就绪型 DAG

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor
from airflow.models import Variable
from datetime import datetime, timedelta
import logging

# 配置日志
logger = logging.getLogger(__name__)

# 从 Airflow Variables 安全读取配置（避免硬编码）
DATA_DIR = Variable.get("data_directory", default_var="/opt/airflow/data")

def extract_data(**context):
    """模拟数据抽取：生成测试文件"""
    import os
    import json
    timestamp = context['ts']
    filepath = f"{DATA_DIR}/raw_{timestamp}.json"
    os.makedirs(os.path.dirname(filepath), exist_ok=True)
    with open(filepath, 'w') as f:
        json.dump({"source": "api", "records": 1000, "ts": timestamp}, f)
    logger.info(f"Extracted data to {filepath}")
    return filepath

def transform_data(**context):
    """模拟数据转换：读取并处理文件"""
    ti = context['ti']
    input_path = ti.xcom_pull(task_ids='extract_task')
    output_path = input_path.replace('raw_', 'transformed_')
    # 简单转换逻辑
    with open(input_path, 'r') as f_in, open(output_path, 'w') as f_out:
        data = json.load(f_in)
        data['processed'] = True
        json.dump(data, f_out)
    logger.info(f"Transformed data saved to {output_path}")
    return output_path

# DAG 默认参数（强制设定，提升健壮性）
default_args = {
    'owner': 'data-engineering-team',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
    'sla': timedelta(hours=1),  # 服务等级协议：DAG 必须在 1 小时内完成
}

# 实例化 DAG（显式定义 schedule_interval，避免隐式行为）
dag = DAG(
    dag_id='etl_pipeline_v2',
    default_args=default_args,
    description='生产环境 ETL 流水线：抽取 → 转换 → 加载',
    schedule_interval='0 2 * * *',  # 每日凌晨 2 点执行（Cron 表达式）
    catchup=False,  # 禁用历史补跑，避免资源冲击
    max_active_runs=1,  # 限制并发运行实例数
    tags=['etl', 'production', 'critical'],
)

# 任务定义（遵循单一职责原则）
start = DummyOperator(task_id='start', dag=dag)
end = DummyOperator(task_id='end', dag=dag)

# 传感器：等待上游数据就绪（生产环境必备）
wait_for_source = FileSensor(
    task_id='wait_for_source_file',
    filepath=f'{DATA_DIR}/upstream_ready.flag',
    poke_interval=300,  # 每 5 分钟检查一次
    timeout=3600,       # 最长等待 1 小时
    mode='poke',
    dag=dag
)

extract = PythonOperator(
    task_id='extract_task',
    python_callable=extract_data,
    provide_context=True,
    dag=dag
)

transform = PythonOperator(
    task_id='transform_task',
    python_callable=transform_data,
    provide_context=True,
    dag=dag
)

load = BashOperator(
    task_id='load_task',
    bash_command=f'echo "Loading transformed data from {DATA_DIR}/transformed_*.json" && '
                 f'ls -la {DATA_DIR}/transformed_*.json',
    dag=dag
)

# 显式声明依赖关系（清晰、可读、可维护）
start >> wait_for_source >> extract >> transform >> load >> end
​

3. 关键高级功能实践

▪ 动态任务生成（适配可变数据源）

# 基于配置动态创建并行处理任务
source_tables = Variable.get("etl_tables", deserialize_json=True, default_var=["users", "orders", "products"])

for table in source_tables:
    # 为每张表生成独立的提取任务
    extract_table_task = PythonOperator(
        task_id=f'extract_{table}',
        python_callable=lambda t=table: extract_data_for_table(t),
        provide_context=True,
        dag=dag
    )
    # 统一汇聚至转换阶段
    extract_table_task >> transform
​

▪ XCom 进阶：跨任务安全数据传递

# 推送结构化数据（避免大对象）
def push_metrics(**context):
    ti = context['ti']
    # 推送轻量元数据
    ti.xcom_push(key='row_count', value=12500)
    ti.xcom_push(key='file_size_mb', value=2.4)

# 拉取并用于条件判断（下游分支）
def validate_quality(**context):
    ti = context['ti']
    row_count = ti.xcom_pull(task_ids='extract_task', key='row_count')
    if row_count < 10000:
        raise ValueError(f"数据量异常：{row_count} < 阈值 10000")
​

▪ 错误处理与可观测性增强

# 集成 Slack 通知（需配置 airflow.cfg 或环境变量）
from airflow.providers.slack.operators.slack import SlackAPIPostOperator

def on_failure_callback(context):
    slack_msg = f"""
    :red_circle: Airflow 任务失败
    *DAG*: {context['dag'].dag_id}
    *任务*: {context['task'].task_id}
    *执行时间*: {context['execution_date']}
    *日志链接*: {context['task_instance'].log_url}
    """
    SlackAPIPostOperator(
        task_id='slack_failure_alert',
        channel='#alerts-data',
        text=slack_msg,
        slack_conn_id='slack_default'
    ).execute(context)

# 在 DAG 中启用
default_args.update({'on_failure_callback': on_failure_callback})
​

生产环境最佳实践指南

• DAG 设计原则
  ✅ 保持 DAG 粒度合理：单个 DAG 应聚焦单一业务域（如 sales_etl、ml_training）
  ✅ 任务命名语义化：fetch_api_v1_orders 优于 task_1
  ✅ 强制设置 sla 与 execution_timeout，防止长任务阻塞调度器

• 安全性与合规性
  ✅ 敏感配置使用 Airflow Connections 或 Variables（启用 Fernet 加密）
  ✅ 禁用 pickle 序列化（enable_xcom_pickling=False），仅允许 JSON 序列化
  ✅ 通过 RBAC 控制 Web UI 访问权限，按角色分配 DataProfiler、User、Op 等角色

• 性能与稳定性
  ✅ 生产环境必用 CeleryExecutor 或 KubernetesExecutor，禁用 SequentialExecutor
  ✅ 定期清理历史任务日志与 XCom 数据（airflow db clean --clean-before-timestamp）
  ✅ 监控关键指标：scheduler_heartbeat, dagbag_import_time, task_instance_duration

• 可维护性
  ✅ DAG 文件遵循 PEP 8，添加类型注解与详细 docstring
  ✅ 使用 airflow dags list、airflow tasks list <dag_id>、airflow dags trigger 进行 CLI 管理
  ✅ 将 DAG 与任务逻辑拆分为独立模块（dags/, operators/, utils/），提升复用性

总结：构建可靠数据基础设施的基石

Apache Airflow 已超越传统调度器范畴，演变为支撑数据驱动决策的核心编排引擎。其核心价值在于：将工作流逻辑从黑盒脚本转化为可版本化、可测试、可审计、可协作的软件资产。通过严谨的 DAG 设计、生产级的错误处理、与云原生生态（Kubernetes、AWS MWAA、GCP Composer）的深度集成，Airflow 赋能团队构建高 SLA、低运维负担、持续演进的数据流水线。掌握其原理与最佳实践，是数据工程师与平台工程师构建现代化数据基础设施的关键能力。