Apache Airflow 数据管道编排实战：DAG 设计、任务依赖与监控告警

Apache Airflow 数据管道编排实战

Apache Airflow 是业界最流行的开源工作流编排平台，被 Airbnb、Shopify、Netflix 等数百家公司用于数据管道调度。本文从核心概念到生产实践，全面讲解 Airflow 的使用方法。

一、Airflow 核心架构

1.1 架构组件

┌─────────────────────────────────────────────────────────────┐
│ Web Server (Flask) │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │
│ │ DAG 列表 │ │ 任务树视图 │ │ 日志与监控 │ │
│ └─────────────┘ └─────────────┘ └─────────────────────┘ │
└────────────────────────┬────────────────────────────────────┘
│ REST API
┌────────────────────────▼────────────────────────────────────┐
│ Scheduler │
│ – 解析 DAG 文件 – 创建 DAG Run │
│ – 确定任务执行顺序 – 任务状态管理 │
└────────────────────────┬────────────────────────────────────┘
│ 消息队列 (Celery / Redis)
┌────────────────────────▼────────────────────────────────────┐
│ Executor │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │
│ │ Local │ │ Celery │ │ Kubernetes │ │
│ │ (开发) │ │ (分布式) │ │ (云原生) │ │
│ └─────────────┘ └─────────────┘ └─────────────────────┘ │
└────────────────────────┬────────────────────────────────────┘
│
┌────────────────────────▼────────────────────────────────────┐
│ Worker │
│ – 实际执行任务 │
│ – 报告任务状态 │
└─────────────────────────────────────────────────────────────┘

1.2 核心概念

| 概念 | 说明 | |——|——| | DAG | 有向无环图，定义工作流的任务和依赖关系 | | Task | DAG 中的单个工作单元 | | Operator | 任务的模板，定义具体执行逻辑 | | DAG Run | DAG 的一次执行实例 | | Task Instance | Task 在特定 DAG Run 中的执行实例 | | Sensor | 特殊 Operator，等待外部条件满足 |

二、环境搭建

2.1 Docker Compose 部署

# docker-compose.yml
version: '3.8'
services:
postgres:
image: postgres:13
environment:
POSTGRES_USER: airflow
POSTGRES_PASSWORD: airflow
POSTGRES_DB: airflow
volumes:
– postgres_data:/var/lib/postgresql/data

redis:
image: redis:7-alpine

airflow-webserver:
image: apache/airflow:2.7.0
command: webserver
ports:
– "8080:8080"
environment:
AIRFLOW__CORE__EXECUTOR: CeleryExecutor
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
AIRFLOW__CELERY__BROKER_URL: redis://redis:6379/0
AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://airflow:airflow@postgres/airflow
volumes:
– ./dags:/opt/airflow/dags
– ./logs:/opt/airflow/logs
depends_on: