4、任务调度引擎：Airflow架构解析、DAG定义与依赖、定时任务配置、任务监控与告警
各位好，我是老赵。今天咱们聊聊任务调度引擎。在金融数据处理里，调度引擎就是整个流水线的“大脑”。你想想看，每天几百个数据源，几十个清洗脚本，十几个模型跑批，谁来决定谁先跑、谁后跑、跑失败了怎么办？
我个人习惯用 Airflow。不是因为它最炫，而是因为它够稳。我在银行做数据中台时，每天凌晨三点跑批，Airflow 从来没掉过链子。嗯，今天我就把它的核心用法掰开揉碎讲给你听。
4.1 Airflow 架构：三个核心组件
Airflow 的架构其实不复杂。说白了就三个东西：Scheduler、Worker、Web Server。
- Scheduler：负责盯着 DAG，看哪个任务该触发了。它就像一个闹钟，到点就响。
- Worker：真正干活的。Scheduler 把任务丢给 Worker，Worker 去执行 Python 脚本、跑 SQL、调 API。
- Web Server：给你看的界面。任务跑没跑完、报没报错，一目了然。
我曾经遇到过一个问题：Scheduler 和 Worker 部署在同一台机器上，结果凌晨跑批时 CPU 打满，整个调度都卡死了。后来我强制把它们分开部署，Scheduler 单独一台轻量机器，Worker 用集群。这个坑，你千万别踩。
4.2 DAG 定义与依赖：把流程画出来
DAG 是 Airflow 的灵魂。它就是一个有向无环图，说白了就是“谁先谁后，不能有死循环”。
我定义一个 DAG 的习惯是这样的：先想清楚数据流，再写代码。比如我要做“日终报表”，流程是：抽取数据 → 清洗 → 计算指标 → 发送邮件。这四个步骤，必须按顺序来。
看代码：
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
def extract():
print("抽取数据...")
def transform():
print("清洗数据...")
def load():
print("加载数据...")
default_args = {
'owner': 'zhao',
'retries': 2,
'retry_delay': timedelta(minutes=5)
}
with DAG(
dag_id='daily_report',
start_date=datetime(2024, 1, 1),
schedule_interval='0 3 * * *', # 每天凌晨3点
catchup=False,
default_args=default_args
) as dag:
t1 = PythonOperator(task_id='extract', python_callable=extract)
t2 = PythonOperator(task_id='transform', python_callable=transform)
t3 = PythonOperator(task_id='load', python_callable=load)
t1 >> t2 >> t3 # 依赖关系：t1完了才t2，t2完了才t3
这里有个细节：catchup=False。我刚开始用 Airflow 时没设这个参数，结果部署一个历史 DAG，它把过去三个月的任务全补跑了，数据库差点炸了。你想想看，凌晨三点突然冒出几百个任务，谁受得了？
4.3 定时任务配置：Cron 表达式与调度策略
定时任务的核心就是 Cron 表达式。Airflow 用的是标准的 5 位 Cron：分 时 日 月 周。
| 场景 | Cron 表达式 | 说明 |
|---|---|---|
| 每天凌晨 3 点 | 0 3 * * * |
日终批处理 |
| 每 30 分钟 | */30 * * * * |
高频数据采集 |
| 每周一早上 8 点 | 0 8 * * 1 |
周报生成 |
| 每月 1 号凌晨 2 点 | 0 2 1 * * |
月结报表 |
我个人习惯把 Cron 表达式写在 DAG 的 schedule_interval 参数里。但要注意：Airflow 的调度时间是基于 start_date 的。比如你设 start_date=2024-01-01，schedule_interval='0 3 * * *'，那么第一个任务会在 2024-01-02 03:00 触发，而不是 1 号当天。
为什么会这样？因为 Airflow 的调度周期是“上一个周期结束后”才触发。这个逻辑一开始容易搞混，我建议你测试时把 start_date 设成昨天，这样马上就能看到效果。
4.4 任务监控与告警：别等用户投诉才发现
任务跑起来了，不代表万事大吉。监控和告警才是保命的。我在金融公司时，每天最怕的就是早上打开手机看到“数据延迟”的投诉。后来我搭了一套告警体系，基本没再出过事。
Airflow 的告警主要靠两个东西：邮件告警和回调函数。
先看邮件告警怎么配：
default_args = {
'email': ['zhao@company.com'],
'email_on_failure': True,
'email_on_retry': False,
'retries': 3,
'retry_delay': timedelta(minutes=5)
}
这样配置后，任务一旦失败，Airflow 会自动发邮件给你。但说实话，邮件有时候不够及时。我更喜欢用 on_failure_callback 对接企业微信或钉钉机器人。
看这个例子：
def send_alert(context):
task_id = context['task_instance'].task_id
dag_id = context['dag'].dag_id
message = f"任务 {dag_id}.{task_id} 失败了！"
# 这里调用企业微信 API 发送消息
print(f"告警发送: {message}")
default_args = {
'on_failure_callback': send_alert
}
我在项目中遇到过一个问题：告警发得太频繁了。一个任务重试 3 次，每次失败都发告警，半夜手机响个不停。后来我加了判断：只在最后一次重试失败时才发告警。代码里判断 context['task_instance'].try_number == context['task_instance'].max_tries 就行。
- 看 DAG 的“运行状态”页面，绿色是成功，红色是失败，黄色是运行中。
- 看“任务持续时间”，如果某个任务突然变慢，说明有问题。
- 看“日志”，Airflow 的日志默认保留 30 天，排查问题全靠它。
嗯，关于 Airflow 的核心内容就这些。调度引擎是整个数据流水线的“心脏”，你把它搞明白了，后面的自动化编排就顺了。记住：架构要稳，DAG 要清，定时要准，告警要快。这四点做到位，你的数据管道基本不会出大问题。