📚 课程目录第4章数据管道构建

4、数据管道构建：使用Apache Airflow等工具编排数据流水线，实现定时、可靠的数据处理

4.1 核心原理：从“脚本定时跑”到“有向无环图编排”

在早期的数据工程实践中，定时数据处理通常依赖 crontab 或 Windows 任务计划程序。这种方式看似简单，却存在致命缺陷：任务之间没有依赖关系管理，前一步失败后后续任务照常执行；没有重试机制；无法直观监控任务状态；更无法处理“数据迟到”“上游分区未就绪”等复杂场景。当数据管道从几条脚本增长到几十个相互依赖的步骤时，cron 脚本会迅速退化为难以维护的“定时炸弹”。

Apache Airflow 的核心思想是将数据管道定义为 有向无环图（DAG）。每个节点（Task）代表一个原子操作，边（Dependency）定义了执行顺序。调度器（Scheduler）根据 DAG 定义的时间表（schedule_interval）触发 DAG Run，执行器（Executor）负责真正运行任务实例。关键设计在于：任务之间通过显式依赖（>> 或 set_downstream）建立关系，Airflow 会确保只有当所有上游任务成功完成后，下游任务才会被触发。若上游失败，下游自动等待或跳过，从而保证数据处理的整体一致性。

核心抽象：DAG（工作流）→ Task（原子操作）→ Operator（执行类型，如 BashOperator、PythonOperator、SparkSubmitOperator）→ Sensor（等待外部条件满足，如文件到达、分区就绪）。

4.2 方法流程：构建一个可运维的 Airflow 数据管道

构建生产级数据管道需要遵循一套标准化的流程，从环境搭建到任务定义，再到部署监控。以下以典型的“每日销售数据同步 + 聚合”场景为例，展示完整实现路径。

4.2.1 环境初始化与目录结构

# 使用官方 Docker Compose 快速搭建 Airflow 2.x 环境
curl -LfO 'https://airflow.apache.org/docs/apache-airflow/2.8.1/docker-compose.yaml'
mkdir -p ./dags ./logs ./plugins ./config
echo -e "AIRFLOW_UID=$(id -u)" > .env
docker compose up -d
# 访问 http://localhost:8080 (admin/admin)

生产环境建议使用 Kubernetes Executor 或 Celery Executor 实现水平扩展。每个 DAG 文件应遵循以下结构：

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor

default_args = {
    'owner': 'data_team',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
    'retry_exponential_backoff': True,
    'max_retry_delay': timedelta(hours=1),
}

with DAG(
    dag_id='sales_etl_daily',
    default_args=default_args,
    description='每日销售数据ETL管道',
    schedule_interval='0 2 * * *',  # 每天凌晨2点
    catchup=False,
    tags=['sales', 'etl'],
) as dag:
    
    wait_for_source = FileSensor(
        task_id='wait_for_source_file',
        filepath='/data/raw/sales/{{ ds }}/sales.csv',
        poke_interval=60,
        timeout=3600,
        mode='poke'
    )
    
    def extract(**context):
        # 实际抽取逻辑
        pass
    
    extract_task = PythonOperator(
        task_id='extract_data',
        python_callable=extract,
        provide_context=True
    )
    
    # 定义依赖
    wait_for_source >> extract_task

4.2.2 关键设计模式

模式	适用场景	实现要点
幂等任务	所有数据处理	使用 `execution_date` 作为分区键，重复执行结果一致
增量抽取	大表同步	记录 last_watermark，只同步变更数据
任务分组	子流程复用	使用 `TaskGroup` 封装可复用的子DAG
动态DAG	多表并行处理	循环生成 Task，注意 `DAG` 解析时间
分支判断	条件执行	`BranchPythonOperator` 根据上游结果选择路径

4.3 案例实战：电商订单数据管道

假设某电商平台需要每天从 MySQL 业务库抽取订单数据，经过清洗后写入 Hive 数仓，并触发下游的报表任务。完整 DAG 定义如下：

from airflow.providers.mysql.hooks.mysql import MySqlHook
from airflow.providers.apache.hive.operators.hive import HiveOperator
from airflow.operators.dummy import DummyOperator

def pull_orders(**context):
    ds = context['ds']
    hook = MySqlHook(mysql_conn_id='mysql_biz')
    sql = f"""
        SELECT order_id, user_id, amount, status, created_at
        FROM orders
        WHERE created_at >= '{ds} 00:00:00'
          AND created_at < DATE_ADD('{ds}', INTERVAL 1 DAY)
    """
    # 拉取并写入临时文件或直接写HDFS
    hook.get_records(sql)

with DAG('order_etl_daily', default_args=default_args,
         schedule_interval='0 3 * * *', catchup=False) as dag:
    
    start = DummyOperator(task_id='start')
    
    pull_mysql = PythonOperator(
        task_id='pull_mysql_orders',
        python_callable=pull_orders
    )
    
    load_hive = HiveOperator(
        task_id='load_to_hive',
        hql="""
            LOAD DATA INPATH '/tmp/orders/{{ ds }}'
            INTO TABLE ods_orders
            PARTITION (dt='{{ ds }}')
        """,
        hive_cli_conn_id='hive_default'
    )
    
    check_data = HiveOperator(
        task_id='check_data_quality',
        hql="""
            SELECT COUNT(*) FROM ods_orders 
            WHERE dt='{{ ds }}' AND amount > 0
        """
    )
    
    start >> pull_mysql >> load_hive >> check_data

最佳实践：使用 execution_date（而非当前时间）作为数据处理的时间维度，确保补数（backfill）时能正确处理历史分区。同时开启 catchup=False 避免误触发大量历史任务。

4.4 常见误区与复盘

在多个企业级项目中，以下问题反复出现，值得重点规避：

误区一：DAG 内做重量级数据计算。Airflow 是编排工具而非计算引擎。将 GB 级数据 join 写在 PythonOperator 中会导致调度器阻塞。正确做法是使用 Spark/ClickHouse 等专用引擎，Airflow 仅负责提交和监控。
误区二：忽略任务幂等性。未按 execution_date 设计分区，导致重跑时数据翻倍或覆盖错误分区。必须保证任意任务可安全重放。
误区三：调度时间与数据可用时间未对齐。例如上游数据源每日 8:00 才生成前一天数据，而管道设置 7:00 运行，导致频繁空跑。应使用 FileSensor 或 ExternalTaskSensor 等待数据就绪。
误区四：依赖外部系统时未设置超时。当 MySQL 慢查询或 HDFS 抖动时，任务可能挂起数小时。必须为每个任务设置 execution_timeout，并配置 on_failure_callback 发送告警。

复盘案例：某金融客户曾因未设置 depends_on_past=True，导致某天上游任务失败后，次日任务仍继续执行，使用前一天不完整的数据生成了错误的风控报表。修复方案：① 为关键链路任务设置 depends_on_past=True；② 增加数据质量校验任务（如行数波动检测）；③ 配置 dag_run_timeout 强制整体超时。

最后，建议为每个 DAG 配置 SLA（如 sla=timedelta(hours=2)），当任务运行超过预期时间时自动告警。同时利用 Airflow 的 Logging 机制将任务日志接入 ELK，便于排查问题。数据管道是数据平台的中枢神经系统，只有做到“定时、可靠、可观测”，才能为上层数据应用提供坚实保障。