标签加工SQL写完了，每天跑一次，但数据延迟了怎么办？

标签加工依赖链路：
 
ODS层（数据采集）    凌晨 01:00 - 02:00
  ods_order_log      -- 订单流水日志
  ods_app_event      -- 埋点行为日志
       |
       v
DWD层（数据清洗）    凌晨 02:00 - 04:00
  dwd_order_detail   -- 清洗后的订单明细（去重、过滤测试数据）
  dwd_user_event     -- 清洗后的行为事件
       |
       v
DWS层（汇总指标）    凌晨 04:00 - 05:30
  dws_user_order_1d  -- 用户每日订单汇总
  dws_user_active_1d -- 用户每日活跃汇总
       |
       v
ADS层（标签输出）    凌晨 05:30 - 07:00
  ads_user_profile   -- 最终的用户画像宽表
       |
       v
同步到Redis/ES        07:00 - 07:30
  推荐系统/运营后台可用

# Airflow DAG示例：标签加工调度（强依赖+弱依赖混合）
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.external_task import ExternalTaskSensor
from datetime import datetime, timedelta
 
dag = DAG('user_profile_daily',
  schedule_interval='0 2 * * *',  # 每天凌晨2点开始
  default_args={'retries': 2, 'retry_delay': timedelta(minutes=10)},
)
 
# 强依赖：等ODS数据就绪
wait_ods = ExternalTaskSensor(
  task_id='wait_ods_ready',
  external_dag_id='ods_ingestion',
  timeout=7200,           # 最多等2小时
  poke_interval=300,      # 每5分钟检查一次
  mode='reschedule',      # 释放Worker槽位
  dag=dag)
 
# DWD清洗 -> DWS汇总 -> ADS标签 -> 同步Redis
dwd_task = BashOperator(task_id='dwd_clean', bash_command='spark-submit dwd.py')
dws_task = BashOperator(task_id='dws_agg', bash_command='spark-submit dws.py')
ads_task = BashOperator(task_id='ads_profile', bash_command='spark-submit ads.py')
sync_redis = BashOperator(task_id='sync_redis', bash_command='spark-submit sync.py')
 
wait_ods >> dwd_task >> dws_task >> ads_task >> sync_redis

数据就绪检查——不能靠猜

很多团队的做法是「上游任务跑完了就触发下游」，但这样有个隐患：上游任务成功了，不代表数据完整了。我在出海公司遇到过这种情况：ODS任务状态是成功的，但实际上只采集了80%的数据（因为有一批海外数据还在路上）。

-- 数据就绪检查：ODS数据量是否达到预期
-- 在启动DWD任务前，先检查ODS数据量
 
-- 方法1：和前N天的平均值对比
WITH today_cnt AS (
  SELECT COUNT(*) AS cnt FROM ods_app_event WHERE dt = '${bizdate}'
),
avg_cnt AS (
  SELECT AVG(cnt) AS avg_cnt FROM (
    SELECT dt, COUNT(*) AS cnt FROM ods_app_event
    WHERE dt BETWEEN DATE_SUB('${bizdate}', 7) AND DATE_SUB('${bizdate}', 1)
    GROUP BY dt
  ) t
)
SELECT
  CASE WHEN t.cnt >= a.avg_cnt * 0.8 THEN 'READY'  -- 达到均值的80%就算就绪
       ELSE 'NOT_READY'
  END AS status
FROM today_cnt t, avg_cnt a;

关键参数：阈值设80%而不是100%，因为数据量天然有波动（周末比工作日少）。如果设100%，周末永远触发不了。也不能设太低（比如50%），否则会在数据严重缺失时仍然跑标签，产出错误的标签值。

面试高频追问

追问一：如果ODS数据到凌晨5点还没到，怎么办？

分两种情况处理。如果是核心标签（比如用户价值等级，运营9点要用），超时2小时后告警通知值班人员，同时启动降级方案——用昨天的标签数据兜底，在报表上标注「数据截至前日」。如果是非核心标签（比如兴趣偏好标签，时效性要求不高），直接跳过当天，等第二天补跑。关键是提前定好降级策略，不要等出问题了再临时决定。

追问二：标签任务重跑怎么保证幂等？

幂等设计非常重要——同一个标签任务跑两次的结果必须一样。我的做法是：标签输出到Hive用INSERT OVERWRITE（而不是INSERT INTO），这样重跑会覆盖而不是追加。输出到Redis用全量覆盖写（SET而不是APPEND）。每个标签任务只处理一天的数据（dt='${bizdate}'），不会影响其他天的数据。这样不管重跑多少次，结果都是一致的。

追问三：你们的标签调度平台用的是什么？

我在不同公司用过不同的调度工具。出海硬件品牌用的是Airflow，DAG定义灵活，Python生态好；短剧出海公司用的是阿里云DataWorks，和MaxCompute深度集成，配置化操作比较方便。如果让我选，中小团队推荐Airflow或DolphinScheduler（国产开源，对中文支持好），大厂一般有自研调度平台。选型核心看团队技术栈——Python团队选Airflow，Java团队选DolphinScheduler。

实战踩坑

我在某短剧出海公司遇到过一次严重的标签延迟事故。某天凌晨4点，海外CDN出了故障，ODS数据到早上8点才全部到齐。但标签调度系统没有做好超时处理——ODS Sensor一直在等，占着Worker不释放，导致其他不依赖这批数据的任务也跑不了（Worker被占满了）。

最终早上10点运营来问数据，不仅标签没出，连日常报表都延迟了。

# 修复后的Sensor配置——超时释放Worker
wait_ods = ExternalTaskSensor(
  task_id='wait_ods_ready',
  external_dag_id='ods_ingestion',
  timeout=7200,           # 最多等2小时就超时
  poke_interval=300,      # 每5分钟检查一次
  mode='reschedule',      # 关键！释放Worker槽位，不阻塞其他任务
  soft_fail=True,         # 超时后标记为skipped而不是failed
  dag=dag)

教训：Sensor任务必须设timeout和mode='reschedule'。默认的mode='poke'会一直占着Worker等待，一旦上游长时间不来，会耗尽Worker资源，影响其他任务。这个坑很隐蔽，平时不出问题，一出就是全线延迟。

总结

标签调度的核心是三件事：依赖管理（DAG编排，强弱依赖分级）、数据就绪检查（不靠猜，用数据量校验）、SLA保障（预留缓冲+告警前置+降级方案）。面试时能讲出调度链路的设计和延迟事故的处理经验，比说「我用了Airflow」有说服力得多。

标签跑出来了，但数据准不准？下一篇聊标签质量监控——100个标签，业务方说不准，怎么排查和保障？

你们标签任务的SLA定在几点？有没有遇到过全线延迟的情况？评论区聊聊～

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