昨天的数据今天才到怎么办——数据就绪检测和Sensor

from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
 
# 等S3上的埋点数据文件到达
wait_for_data = S3KeySensor(
    task_id='wait_for_events_data',
    bucket_name='data-lake',
    bucket_key='raw/events/dt={{ ds }}/_SUCCESS',  # 检查完成标志
    timeout=7200,          # 最多等2小时
    poke_interval=300,     # 每5分钟检查一次
    mode='reschedule',     # 等待时释放Worker！
    soft_fail=True,        # 超时标记skipped不是failed
)
 
 # _SUCCESS文件是上游写完数据后创建的标志文件
# 有_SUCCESS说明数据完整，可以开始ETL
# 没有_SUCCESS说明上游还没写完或写入失败
 
wait_for_data >> ods_load_events  # 数据到了才加载

SqlSensor——等数据行数达标

from airflow.providers.common.sql.sensors.sql import SqlSensor
 
# 等源数据库的数据量达到预期
wait_for_rows = SqlSensor(
    task_id='wait_for_source_ready',
    conn_id='mysql_source',
    sql="""
        SELECT COUNT(*) > 1000000
        FROM raw_events
        WHERE dt = '{{ ds }}'
    """,
    timeout=3600,
    poke_interval=600,     # 每10分钟查一次
    mode='reschedule',
)
 
 # 返回True（行数>100万）时Sensor标记成功
# 返回False时继续等待
# 适合检测上游数据库的ETL是否完成

_SUCCESS标志文件模式

大数据生态中最常用的数据就绪标志是_SUCCESS文件：

工作方式：上游作业写完所有数据后在目录中创建一个名为_SUCCESS的空文件。下游Sensor检测_SUCCESS文件是否存在来判断数据是否就绪。

为什么不直接检查数据文件？因为上游可能还在写入中——文件存在但数据不完整。_SUCCESS文件保证了'上游认为数据已经完整写入了'。Spark/Hive的INSERT OVERWRITE默认会在输出目录创建_SUCCESS文件。

实战经验：S3数据延迟的处理方案

在某短视频平台，S3上的埋点数据通常0:30到齐但偶尔延迟到3:00。我的处理方案：

正常情况：Sensor在1:00开始检测，0:30数据已到，Sensor立刻通过，ETL正常执行。

轻微延迟（到2:00）：Sensor持续检测，2:00数据到达后通过，ETL延迟1小时但SLA（4:30前完成）还能满足。

严重延迟（超过3:00）：Sensor在3:00超时（timeout=2h，从1:00开始），标记skipped（soft_fail=True）。ODS加载任务跳过这个数据源，用前一天的数据兜底。同时触发P1告警通知值班人排查上游延迟原因。

效果：即使上游数据延迟3小时以上，核心报表最多延迟到3:00就能用旧数据兜底产出。之前如果不设超时Sensor会一直等，可能等到早上8点都出不了报表。

Deferrable Operators——Sensor的进化

传统Sensor即使用reschedule模式也需要定期占用Worker Slot做检查。Airflow 2.2+的Deferrable Operators彻底解决了这个问题：

原理：Deferrable Operator执行后把自己'挂起'到Triggerer组件，完全释放Worker Slot。Triggerer用异步IO同时监控成百上千个挂起的任务。条件满足时Triggerer唤醒Task，Task再申请Worker Slot继续执行。

好处：100个Sensor同时等待只需要1个Triggerer进程（异步IO），而不需要100个Worker Slot。对于大规模调度（几百个Sensor）显著节省资源。

面试高频追问

追问一：poke模式和reschedule模式区别？

poke模式Sensor持续占着Worker Slot——检查失败就sleep然后再检查循环。reschedule模式每次检查后释放Slot，到下次poke_interval再申请Slot。生产必须用reschedule。poke只适合短时间（几分钟）的等待。

追问二：如果上游永远不到怎么办？

timeout+soft_fail。Sensor超时后标记skipped——下游根据trigger_rule决定是用旧数据兜底还是也跳过。同时告警通知值班人排查上游系统。永远不要让Sensor无限等待——必须有超时兜底。

追问三：DataWorks怎么做数据就绪检测？

DataWorks没有Sensor概念，但有'跨周期依赖'——任务可以配置依赖'上一周期的某个节点完成'。对于外部数据到达检测，DataWorks用'数据集成+调度触发'——数据集成任务完成后自动触发下游。没有Airflow Sensor那么灵活但够用。

总结

Sensor面试讲三层：① Sensor的作用和类型（S3/HDFS/SQL/ExternalTask等，持续检测条件满足才让下游执行）；② 必须设timeout+soft_fail（防止无限等待，超时用旧数据兜底）；③ mode选择（必须用reschedule或Deferrable，不要poke浪费Worker）。能讲出S3数据延迟的三级处理方案（正常/轻微延迟/严重延迟），面试官觉得你设计的调度系统有完善的容错能力。

下一篇聊Backfill回溯：要补最近30天的数据怎么办？

你们怎么检测上游数据是否就绪？评论区聊聊～

关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长