📚 课程目录第4章调度周期与触发

4. 调度周期与触发：Cron表达式、事件驱动、传感器(Sensor)模式

调度周期和触发方式，说白了就是决定「你的流水线什么时候跑」。很多新手上来就写一个 Cron 表达式，觉得能跑就行。但我在项目中见过太多次因为触发时机不对导致的数据灾难——要么跑太频繁把数据库打崩，要么跑太慢让业务等得跳脚。

这一章，我把三种主流触发方式掰开揉碎讲清楚。你想想看，搞懂了它们，你就能精准控制每一段流水线的「呼吸节奏」。

4.1 Cron表达式：最经典的定时调度

Cron 表达式是 Linux 世界的遗产，也是调度系统的基石。它用一串字符描述「什么时候执行」。我个人习惯把它拆成 5 段（标准 Unix Cron）或 6 段（带秒的 Quartz 风格）。

字段	取值范围	特殊字符
秒（可选）	0-59	, - * /
分	0-59	, - * /
时	0-23	, - * /
日	1-31	, - * ? / L W
月	1-12 或 JAN-DEC	, - * /
周	0-7 或 SUN-SAT	, - * ? / L #

举个例子：

# 每天凌晨 2:30 执行
30 2 * * *

# 每 5 分钟执行一次
*/5 * * * *

# 每月 1 号和 15 号的上午 9 点执行
0 9 1,15 * *

# 工作日（周一至周五）的晚上 8 点执行
0 20 * * 1-5

我曾经踩过的坑： 用 Cron 做小时级调度时，没考虑夏令时。结果每年 3 月某天凌晨 2 点那一次调度直接「消失」了，导致那小时的数据没处理。后来我强制所有 Cron 表达式都使用 UTC 时间，再在应用层做时区转换。

Cron 适合什么场景？固定周期的批处理任务，比如日报、小时级数据同步、定时清理日志。但它有个硬伤——它不知道上游数据有没有准备好。你设了 8 点跑，如果上游 8 点 05 分才把数据推过来，那这次调度就白跑了。

4.2 事件驱动：让数据自己「喊」你

事件驱动，说白了就是「数据到了你再跑」。这种方式比 Cron 更智能，因为它能感知外部变化。我在做实时数仓时，几乎全部采用事件驱动。

常见的事件源包括：

文件到达事件：HDFS/S3 上有新文件写入
消息队列事件：Kafka/RabbitMQ 中有新消息
API 回调：上游系统通过 Webhook 通知你
数据库变更：CDC（Change Data Capture）捕获到表变化

在 Airflow 中，事件驱动通常通过 TriggerDagRunOperator 或 ExternalTaskSensor 实现。举个例子：

from airflow.operators.trigger_dagrun import TriggerDagRunOperator

# 当上游 DAG 成功完成后，触发下游 DAG
trigger = TriggerDagRunOperator(
    task_id='trigger_downstream',
    trigger_dag_id='data_processing_dag',
    execution_date='{{ ds }}',
    reset_dag_run=True,
    wait_for_completion=True,
)

我的建议： 事件驱动虽然灵活，但别滥用。如果每个小文件都触发一次调度，你的调度器会被「事件风暴」冲垮。我一般会加一个「事件聚合窗口」——比如 5 分钟内的事件合并成一次触发。

4.3 传感器(Sensor)模式：等，也是一种策略

传感器模式，是事件驱动的一种特殊实现。它像一个「哨兵」，不断轮询某个条件是否满足，直到条件成立才放行下游任务。

你想想看，为什么需要 Sensor？因为很多时候，数据不是「瞬间」准备好的。比如：

等待某个 HDFS 目录下出现指定数量的文件
等待 SQL 查询返回非空结果
等待外部 API 返回特定状态码
等待前一天的分区数据写入完成

Airflow 内置了很多 Sensor：

Sensor 名称	用途	常用参数
`FileSensor`	等待文件或目录出现	filepath, fs_conn_id
`ExternalTaskSensor`	等待另一个 DAG 的任务完成	external_dag_id, external_task_id
`SqlSensor`	等待 SQL 查询返回结果	conn_id, sql, success
`HdfsSensor`	等待 HDFS 文件出现	filepath, hdfs_conn_id
`S3KeySensor`	等待 S3 对象出现	bucket_name, bucket_key

一个典型的 Sensor 用法：

from airflow.sensors.filesystem import FileSensor

# 等待 /data/input/ 目录下出现 .csv 文件
wait_for_file = FileSensor(
    task_id='wait_for_input_file',
    filepath='/data/input/*.csv',
    fs_conn_id='local_fs',
    poke_interval=30,  # 每 30 秒检查一次
    timeout=3600,      # 最多等 1 小时
    mode='reschedule', # 不占用 worker 资源
)

  核心区别： Cron 是「到点就干」，事件驱动是「有事才干」，Sensor 是「等事干了再干」。三者在实际项目中经常组合使用。

4.4 三种模式的组合策略

在实际项目中，我很少只用一种模式。比如一个典型的数据流水线可能是这样的：

Cron 定时启动：每天 8 点，调度器启动一个「数据检查 DAG」
Sensor 等待条件：检查上游系统是否在 8 点前推送了数据文件
事件驱动触发：文件到达后，触发下游的数据清洗和加载任务
再嵌套 Sensor：清洗任务完成后，等待数据仓库的分区就绪，再触发报表生成

这种组合方式，既保证了定时任务的稳定性，又保留了事件驱动的灵活性。嗯，这里要注意：组合使用时要小心「死锁」。比如两个 DAG 互相等待对方完成，那整个调度就卡死了。我一般会在 Sensor 上加 timeout 参数，超时后发告警而不是无限等待。

4.5 知识体系总览

下面这张图，把三种触发方式的关系和适用场景串起来了。你可以把它当作本章的「思维导图」。

4.6 避坑指南

最后，我把这些年踩过的坑总结一下。你如果正在设计调度系统，这几条能帮你省不少时间。

我曾经犯过的错：

Cron 时区陷阱：服务器时区是 UTC，业务时区是 CST，结果所有定时任务都偏了 8 小时。后来我统一在 Cron 表达式里写 UTC，再在 DAG 参数里声明 timezone。
Sensor 轮询太频繁：默认 poke_interval=60 没问题，但有人改成 5 秒，结果 100 个 Sensor 同时跑，数据库连接池被打满。我建议至少 30 秒起步。
事件驱动丢失：Kafka 消息被消费但处理失败，又没有重试机制，数据就丢了。我现在所有事件驱动都配死信队列和重试策略。
Sensor 死锁：DAG A 等 DAG B 完成，DAG B 等 DAG A 完成，两个都卡死。解决方案是加超时和告警，或者用 ExternalTaskSensor 的 allowed_states 参数放宽条件。

嗯，调度触发这块，说白了就是「什么时候干活」的问题。Cron 是闹钟，事件驱动是门铃，Sensor 是哨兵。搞懂这三者的区别和组合方式，你的流水线就能像瑞士钟表一样精准运转。