20、Airflow实战：ETL DAG编写、依赖管理、监控与告警、与Docker集成
说实话，Airflow 这东西刚出来那会儿，我第一反应是「又一个调度工具？」。但真正用起来才发现，它把 ETL 的编排、依赖、重试、监控全揉到一起了。今天咱们就聊聊怎么用它写 DAG、管依赖、配告警，再把它塞进 Docker 里跑起来。
20.1 先画张图，理清 Airflow 的核心逻辑
我个人习惯，学任何工具先画一张逻辑图。Airflow 的核心就三块：DAG 定义、调度器、执行器。下面这张 SVG 图帮你快速建立整体认知。
你看，整个流程其实不复杂。DAG 定义好任务依赖，调度器按顺序触发，执行器真正干活。监控和 Docker 是外围的保障手段。下面咱们一个一个拆开讲。
20.2 编写第一个 ETL DAG
写 DAG 说白了就是写 Python 代码。我习惯把 DAG 定义和任务逻辑分开，这样好维护。来看一个实际的 ETL 例子：每天从 API 拉数据，清洗后写入数据仓库。
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.dummy import DummyOperator
default_args = {
'owner': 'etl_team',
'depends_on_past': False,
'email_on_failure': True,
'email': ['alerts@company.com'],
'retries': 2,
'retry_delay': timedelta(minutes=5),
}
with DAG(
dag_id='etl_daily_pipeline',
default_args=default_args,
description='每日 ETL：API 抽取 -> 清洗 -> 加载',
schedule_interval='0 2 * * *', # 每天凌晨2点
start_date=datetime(2024, 1, 1),
catchup=False,
tags=['etl', 'production'],
) as dag:
start = DummyOperator(task_id='start')
def extract_from_api(**context):
# 实际项目中这里调用 requests 拉数据
print("抽取数据完成")
return {"status": "ok", "records": 1000}
def transform_data(**context):
# 数据清洗、去重、格式转换
print("转换数据完成")
def load_to_dw(**context):
# 写入 PostgreSQL / ClickHouse 等
print("加载数据完成")
extract = PythonOperator(
task_id='extract_from_api',
python_callable=extract_from_api,
provide_context=True,
)
transform = PythonOperator(
task_id='transform_data',
python_callable=transform_data,
provide_context=True,
)
load = PythonOperator(
task_id='load_to_dw',
python_callable=load_to_dw,
provide_context=True,
)
end = DummyOperator(task_id='end')
# 依赖关系：start >> extract >> transform >> load >> end
start >> extract >> transform >> load >> end
我的小习惯：每个任务都加上 retries 和 retry_delay。网络抖动是常态，重试能省去半夜爬起来看告警的麻烦。我曾经就因为没配重试，凌晨三点被电话叫醒——就为了一个临时网络超时。
20.3 依赖管理：不只是「>>」这么简单
你想想看，实际 ETL 场景里依赖关系复杂得多。比如：
- 分支依赖：数据源 A 和 B 都成功才能继续
- 条件触发：只有昨天数据完整才跑今天的任务
- 跨 DAG 依赖：上游 DAG 跑完才触发下游 DAG
Airflow 提供了几种方式来处理这些场景。我个人最常用的是 ExternalTaskSensor 和 TriggerDagRunOperator。
20.3.1 跨 DAG 依赖
from airflow.sensors.external_task import ExternalTaskSensor
wait_for_upstream = ExternalTaskSensor(
task_id='wait_for_upstream_dag',
external_dag_id='data_ingestion_dag',
external_task_id='load_complete',
timeout=3600,
poke_interval=300, # 每5分钟检查一次
mode='reschedule',
)
# 上游数据加载完成后再执行本 DAG 的 extract
wait_for_upstream >> extract
注意：ExternalTaskSensor 默认检查 execution_date 完全一致。如果你的 DAG 调度时间不同，记得设置 execution_delta 参数。我曾经踩过这个坑，查了半天才发现是日期对不上。
20.3.2 条件分支与短路
有时候某个任务失败了，下游任务不应该继续执行。Airflow 默认行为是「上游失败，下游跳过」。但如果你想让下游执行一个「补偿任务」，可以用 BranchPythonOperator。
from airflow.operators.python import BranchPythonOperator
def check_data_quality(**context):
# 检查数据质量，返回下一步要执行的任务 ID
if context['ti'].xcom_pull(task_ids='extract_from_api')['records'] > 0:
return 'transform_data'
else:
return 'send_alert'
branch = BranchPythonOperator(
task_id='check_quality',
python_callable=check_data_quality,
provide_context=True,
)
extract >> branch
branch >> [transform, send_alert]
# 如果走 send_alert 分支，transform 会被自动跳过
20.4 监控与告警：别等用户发现数据不对
说实话，ETL 工程师最怕的就是「数据静默错误」——任务显示成功，但数据是错的。所以监控必须做到三层：
| 监控层级 | 监控内容 | 告警方式 |
|---|---|---|
| 任务级 | 任务失败、超时、重试次数 | Email / Slack 即时通知 |
| 数据级 | 数据量波动、空值率、延迟 | 自定义检查 + 告警 |
| 平台级 | 调度器健康、队列积压、资源使用 | PagerDuty / 自建看板 |
20.4.1 任务级告警配置
在 default_args 里配好 email_on_failure 是最基本的。但我更推荐用 Slack Webhook，响应更快。
from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator
def send_slack_alert(context):
slack_msg = f"""
:red_circle: DAG 失败告警
DAG: {context['dag'].dag_id}
Task: {context['task'].task_id}
Execution Time: {context['execution_date']}
Log URL: {context['task_instance'].log_url}
"""
slack_alert = SlackWebhookOperator(
task_id='slack_alert',
webhook_token='your_slack_webhook_token',
message=slack_msg,
channel='#etl-alerts',
)
return slack_alert.execute(context=context)
# 在 DAG 中绑定告警回调
dag.on_failure_callback = send_slack_alert
避坑指南：我曾经把 Slack Token 直接写在代码里，结果不小心提交到了 Git 仓库。后来改用 Airflow 的 Connection 或 Variables 来存储敏感信息，安全多了。
20.4.2 数据质量监控
任务成功了不代表数据没问题。我习惯在每个 DAG 末尾加一个数据质量检查任务。
def check_row_count(**context):
# 假设从数据库查询今天加载的行数
row_count = query_database("SELECT COUNT(*) FROM dw_table WHERE dt = CURRENT_DATE")
expected_min = 10000
if row_count < expected_min:
raise ValueError(f"数据量异常：预期 {expected_min} 行，实际 {row_count} 行")
print(f"数据质量检查通过，共 {row_count} 行")
quality_check = PythonOperator(
task_id='data_quality_check',
python_callable=check_row_count,
provide_context=True,
)
load >> quality_check >> end
20.5 与 Docker 集成：环境一致性的终极方案
你想想看，开发环境跑得好好的，一上生产就报错——这种事我见得太多了。Docker 能完美解决这个问题。Airflow 官方提供了 DockerOperator，可以直接在 DAG 里启动容器执行任务。
20.5.1 使用 DockerOperator
from airflow.providers.docker.operators.docker import DockerOperator
etl_task = DockerOperator(
task_id='run_etl_in_container',
image='my_etl_image:latest',
api_version='auto',
auto_remove=True,
docker_url='unix://var/run/docker.sock',
network_mode='bridge',
environment={
'DB_HOST': 'postgres_host',
'DB_PORT': '5432',
},
mounts=[
'/data/etl_output:/output', # 挂载宿主机目录
],
command='python /app/run_etl.py --date {{ ds }}',
)
start >> etl_task >> end
注意：DockerOperator 需要 Airflow 所在机器能访问 Docker 守护进程。如果你用的是 CeleryExecutor，每个 Worker 节点都要安装 Docker。我建议用 Docker Compose 或 Kubernetes 来统一管理 Airflow 本身和任务容器。
20.5.2 用 Docker Compose 部署 Airflow
本地开发时，我推荐用官方提供的 docker-compose.yaml。一行命令就能拉起整套环境：
# 下载官方 docker-compose 文件
curl -LfO 'https://airflow.apache.org/docs/apache-airflow/stable/docker-compose.yaml'
# 初始化数据库
docker-compose up airflow-init
# 启动所有服务
docker-compose up -d
这个 compose 文件包含了：
- PostgreSQL（元数据库）
- Redis（消息队列，用于 CeleryExecutor）
- Airflow Webserver（UI 界面）
- Airflow Scheduler（调度器）
- Airflow Worker（执行器，可扩展）
我个人习惯把 DAG 文件放在 ./dags 目录，日志放在 ./logs，插件放在 ./plugins。这样即使容器重启，数据也不会丢。
20.5.3 生产环境的最佳实践
嗯，这里要注意一点：生产环境别直接用 Docker Compose。我建议用 Kubernetes + Helm Chart 来部署 Airflow。Helm 官方 Chart 已经封装好了所有组件，支持自动扩缩容、滚动更新、持久化存储。
如果你团队规模不大，用 Docker Compose 加几台机器也够用。关键是把 DAG 代码和镜像版本管理好，每次发布新 DAG 时重新构建镜像。
核心总结：Airflow 的 DAG 编写要注重依赖清晰、重试合理；监控要覆盖任务级和数据级；Docker 集成能保证环境一致性，减少「在我机器上能跑」的尴尬。把这三点做好，你的 ETL 管道就能稳定运行了。
公众号:蓝海资料掘金营，微信deep3321