数据采集任务监控——延迟、丢失、重复指标怎么打？

数据采集手册第 16 题约 11 分钟读完

本文目录

先说结论
业务场景：静默漏数据被业务方发现
方案一：延迟（Latency）监控
方案二：完整率（Completeness）监控
方案三：准确率（Accuracy）监控
指标体系对比
面试高频追问
实战踩坑：磁盘满导致Checkpoint失败但作业继续跑
总结

先说结论

监控三大核心指标：Latency（延迟）、Completeness（完整率）、Accuracy（准确率）。少一个都不完整。

指标分层打点：采集侧打"任务级"指标、数据侧打"事件级"指标、业务侧打"结果级"指标，三层对账才能发现隐藏问题。

告警必须分级：P0核心库延迟、P1任务失败、P2数据质量异常。不分级=不告警，所有告警都响=没人看。

先把这段讲顺；面试官追问时，再展开下面的原理和细节。

详细讲解

数据采集最怕"静默失败"——任务在跑但数据少了、重复了、延迟了，下游报表出错了才发现。一个成熟的数据团队，采集监控的覆盖度直接决定了数据质量。这道题面试官想看的不是"加告警"，而是你有没有完整的可观测性体系——从指标定义到告警分级到故障定位的闭环。

业务场景：静默漏数据被业务方发现

有一次业务方跑报表发现前一天某个渠道的订单数少了一半，反馈上来数据团队才开始排查，结果是上游Kafka一个partition的消费者因bug假死了12小时，监控系统里看一切"正常"——任务在跑、没报错。这次事件直接让团队把监控从"任务级"升级到"数据级"。

故障现场复盘:
  09:00  Consumer一个线程因NullPointerException进入死循环
         (没崩溃只是卡住)
  09:00~21:00  该partition数据堆积,但作业JVM活着,心跳正常
         Grafana上: CPU正常/内存正常/任务状态RUNNING
         下游Kafka lag监控只盯了整个topic的lag,平均到8个partition不明显
  21:00  业务报表对账发现渠道X订单少一半
  21:30  人工查Kafka每个partition的lag,发现partition-3有12小时lag
 
  教训: 监控太粗,掩盖了单partition的故障

方案一：延迟（Latency）监控

延迟是采集监控最基础的指标。不是"作业跑完"的延迟，而是"事件产生到落库"的端到端延迟。

// 端到端延迟打点: 每个事件带时间戳,每一环都打lag
// 1) 客户端/源头: event.produce_ts
// 2) Kafka入口:  event.kafka_enter_ts  (Producer侧打)
// 3) 流处理:    event.process_ts     (Flink算子打)
// 4) Sink入库:  event.sink_ts         (写入时记录)
 
// Flink算子里上报metric
public class LatencyMonitor extends RichMapFunction<Event, Event> {
    private transient Histogram produceToSinkLatency;
    
    public void open(Configuration cfg) {
        produceToSinkLatency = getRuntimeContext()
            .getMetricGroup().histogram("produce_to_sink_latency_ms",
                new DescriptiveStatisticsHistogram(10000));
    }
    
    public Event map(Event e) {
        long now = System.currentTimeMillis();
        produceToSinkLatency.update(now - e.getProduceTs());
        // 按来源/事件类型/topic打标签,方便分维度分析
        getRuntimeContext().getMetricGroup()
            .addGroup("source", e.getSource())
            .gauge("event_age_ms", () -> now - e.getProduceTs());
        return e;
    }
}
 
// Prometheus查询: 按分位数看延迟分布
// histogram_quantile(0.99, rate(flink_task_produce_to_sink_latency_ms_bucket[1m]))
// 告警: p99 > 60000ms 持续3分钟 → P1

适用场景：所有流式采集必打。离线采集用"任务完成时间 - 数据业务时间"算延迟。

方案二：完整率（Completeness）监控

完整率是最容易漏的指标——"任务成功了"不等于"数据都在"。必须独立打"应该有多少条"和"实际拉到多少条"，两头对账。

-- 源端记录每小时的数据量(由业务系统上报,或数据团队定时拉取)
CREATE TABLE source_rowcount_expected (
  source_name STRING,  -- 源表名
  hour_ts     BIGINT,   -- 小时时间戳
  expected_rows BIGINT  -- 源端该小时的实际行数
);
 
-- 下游统计
CREATE TABLE sink_rowcount_actual (
  source_name STRING,
  hour_ts     BIGINT,
  actual_rows BIGINT   -- 采集到数仓的行数
);
 
-- 每小时对账(告警SQL)
SELECT
  e.source_name, e.hour_ts,
  e.expected_rows, a.actual_rows,
  ROUND(100.0 * a.actual_rows / e.expected_rows, 2) AS completeness_pct
FROM source_rowcount_expected e
LEFT JOIN sink_rowcount_actual a
  ON e.source_name = a.source_name AND e.hour_ts = a.hour_ts
WHERE e.hour_ts = UNIX_TIMESTAMP(DATE_SUB(NOW(), INTERVAL 1 HOUR))
  AND (a.actual_rows IS NULL OR a.actual_rows < e.expected_rows * 0.99);
-- 完整率 < 99% 告警

关键点：完整率不是"近似值"——必须有源端权威数字。对MySQL CDC场景，可以用"SELECT COUNT(*) FROM source_table"定时对账；对埋点，用会话级别的元事件对账（见前面讲的埋点SDK章节）。

方案三：准确率（Accuracy）监控

准确率是最深层的监控——数据到了但字段错了、类型转换出问题、枚举值不对。这类问题不影响行数，但业务分析就是会错。

-- 数据质量规则库(DQC: Data Quality Check)
CREATE TABLE dq_rules (
  rule_id    STRING,
  table_name STRING,
  rule_type  STRING,  -- not_null/range/enum/regex/unique/reference
  rule_body  STRING,  -- SQL表达式
  severity   STRING   -- P0/P1/P2
);
 
-- 示例规则:
-- 1) user_id 非空率 >= 99.99%
-- 2) amount 在 [0.01, 1000000] 范围内
-- 3) status 必须在 enum('created','paid','cancelled','refunded')
-- 4) phone 符合正则 ^1[3-9]\d{9}$
-- 5) user_id 在 dim_users 中存在(外键一致性)
 
-- 每小时跑一次DQC
INSERT INTO dq_results
SELECT
  '${rule_id}' AS rule_id, NOW() AS check_time,
  (SELECT COUNT(*) FROM ods_orders WHERE ds = '${bizdate}') AS total_rows,
  (SELECT COUNT(*) FROM ods_orders WHERE ds = '${bizdate}'
     AND (user_id IS NULL OR user_id = '')) AS violation_rows,
  ROUND(100.0 * violation_rows / total_rows, 4) AS violation_rate;
-- violation_rate > 0.01% 且规则severity=P0 → 告警

指标体系对比

维度	Latency	Completeness	Accuracy
监控层面	任务级	数据量级	字段级
打点成本	低(算子metric)	中(对账作业)	高(DQC规则)
发现问题类型	卡死/积压	漏采/丢失	格式/口径错
告警阈值	p99延迟	完整率%	违反率%
必要性	所有任务必打	核心任务必打	关键数据必打

面试高频追问

追问一：告警太多都没人看怎么办？

典型的"告警疲劳"问题，靠三招治理：① 分级（P0/P1/P2对应不同通知渠道——P0电话、P1钉钉@、P2邮件）；② 降噪（相同告警合并、连续触发才发、设置静默窗口）；③ 自动恢复标识（告警恢复时也发通知，避免"到底好没好"的困扰）。最关键的是：定期review告警有效性，连续1个月没触发的告警要么删要么调阈值。告警不是越多越好，是越准越好。

追问二：Flink作业的监控怎么打全？

Flink自带的metrics很全，重点盯四组：① Checkpoint——duration、size、failure count，失败率>5%告警；② Backpressure——反压比例超过50%说明瓶颈在下游；③ Lag——Kafka consumer lag、source的watermark延迟；④ 业务指标——自己打的numRecordsIn/Out、自定义histogram。生产上把这些指标接入Prometheus，用Grafana建标准看板，新作业直接套模板，不用每次从头配。

追问三：数据血缘对监控有什么用？

血缘是"智能告警"的基础。当ODS层出问题时，能自动推算出影响到哪些DWD/DWS/报表，一次告警覆盖全链路。同时blame定位也方便——下游报表出错时，从血缘往上追一层层查，比盲查快10倍。生产上血缘数据可以从SQL静态解析（如Atlas/OpenLineage）或运行时捕获（Flink algorithm血缘插件）。

实战踩坑：磁盘满导致Checkpoint失败但作业继续跑

有次Flink作业所在的TaskManager磁盘满了，Checkpoint持续失败，但作业本身没挂（Flink默认checkpoint失败不影响数据流转），结果连续12小时checkpoint全部超时，一旦作业重启就得回放12小时的Kafka。幸好这段时间没重启，但风险极大。排查发现告警只监控了"作业状态"没监控"checkpoint"。

# ❌ 反模式: 只看作业状态
# flink_job_status{job="xxx", status="RUNNING"} → 看似OK
 
# ✓ 必须监控的Checkpoint关键指标
# 1) Checkpoint连续失败
flink_jobmanager_job_numberOfFailedCheckpoints - 
  flink_jobmanager_job_numberOfFailedCheckpoints offset 10m > 5
  → 10分钟内连续5次失败 P1告警
 
# 2) Checkpoint时间超过interval
flink_jobmanager_job_lastCheckpointDuration > 300000
  → 超过5分钟 P2告警
 
# 3) 最后一次成功checkpoint距今太久
(time() - flink_jobmanager_job_lastCheckpointRestoreTimestamp/1000) > 1800
  → 30分钟没成功过 P0告警(重启风险)
 
# 4) TM磁盘使用率
node_filesystem_avail_bytes / node_filesystem_size_bytes < 0.1
  → 磁盘<10%可用 P1告警

解决方案：把Checkpoint失败率、最后成功时间、TM磁盘使用率都纳入监控，配P0-P2分级告警。Flink的"优雅失败"特性既是优点也是坑，必须主动监控那些"不致命但有风险"的状态。从此之后所有流作业都跑这个标准看板。

总结

采集监控的体系化思路是"三指标+三层对账+三级告警"：Latency/Completeness/Accuracy三指标覆盖、任务级/数据级/业务级三层独立打点对账、P0/P1/P2三级告警避免疲劳。核心不是"能发现问题"，而是"能快速定位和恢复"。面试时讲"可观测性闭环"比讲单点metric分高得多。

下一篇聊Exactly-Once的老话题：采集到的数据有重复，Exactly-Once到底能不能保证？

你们采集系统的监控覆盖度多少？告警多不多？评论区聊聊～

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

刷题进度保存在当前浏览器