4. 监控与告警体系：监控指标采集、告警阈值设定、告警通知策略
监控告警体系，说白了就是数据管道的「仪表盘」和「警报器」。没有它，你的管道跑得好不好全靠玄学。我见过太多团队，管道崩了半小时才被用户投诉发现——那场面，真的尴尬。
这一章，我们聊聊怎么把监控告警做到位。从指标采集到阈值设定，再到通知策略，一条线串起来。
4.1 监控指标采集：你得知道看什么
采集指标之前，先想清楚一个问题：什么数据能反映管道的真实健康状态？
我个人习惯把指标分成四类，缺一不可：
| 指标类别 | 典型指标 | 采集方式 |
|---|---|---|
| 吞吐量指标 | 每秒处理行数、每秒写入字节数 | Prometheus Client 埋点 |
| 延迟指标 | 端到端延迟、单阶段处理耗时 | 日志打点 + 分布式追踪 |
| 数据质量指标 | 空值率、重复率、Schema 校验失败数 | 数据校验框架上报 |
| 资源消耗指标 | CPU 使用率、内存占用、连接池水位 | Node Exporter / JMX |
我的经验：别一上来就采集上百个指标。先抓核心的 5-8 个，跑通后再扩展。我曾经在一个项目里采集了 50 多个指标，结果告警配置写到崩溃，最后发现 80% 的问题靠 3 个指标就能发现。
采集频率怎么定？我建议：
- 吞吐量和延迟：每 15-30 秒采集一次，太密会增加开销，太疏会漏掉毛刺
- 数据质量：每个批次完成后采集，或者每 5 分钟聚合一次
- 资源消耗：每 60 秒采集一次，变化相对平缓
下面是一个简单的 Prometheus 埋点示例，我用在 Flink 任务里：
// 自定义 Counter 统计处理行数
val processedRows = Metrics.counter("pipeline_rows_processed_total")
processedRows.inc()
// Gauge 记录当前延迟（毫秒）
val currentLatency = Metrics.gauge("pipeline_latency_ms")
currentLatency.set(latencyMs)
// Histogram 记录批次处理耗时
val batchDuration = Metrics.histogram("pipeline_batch_duration_seconds")
batchDuration.observe(durationSeconds)4.2 告警阈值设定：别让告警变成噪音
阈值设得太松，问题漏报；设得太紧，告警疲劳。你想想看，每天收到几百条告警，最后谁还会认真看？
我总结了一套「三层阈值法」，实战中很管用：
- 警告阈值（Warning）：指标偏离基线 20%，持续 5 分钟。这时候管道还能跑，但需要关注。
- 严重阈值（Critical）：指标偏离基线 50%，持续 2 分钟。必须立即处理，否则要出大事。
- 灾难阈值（Disaster）：指标直接归零或爆表，持续 30 秒。管道已经挂了，全员响应。
核心原则：阈值不是拍脑袋定的。先跑两周基线数据，取 P95 和 P99 作为参考。我习惯用「动态基线」——让系统根据历史数据自动调整阈值，而不是写死一个数字。
举个例子，端到端延迟的阈值设定：
| 级别 | 阈值 | 持续时长 | 响应要求 |
|---|---|---|---|
| Warning | 延迟 > 基线 P95 × 1.2 | ≥ 5 分钟 | 值班人员 30 分钟内确认 |
| Critical | 延迟 > 基线 P99 × 1.5 | ≥ 2 分钟 | 值班人员 10 分钟内介入 |
| Disaster | 延迟 > 基线 P99 × 3 或 无数据 | ≥ 30 秒 | 全员响应，5 分钟内上线 |
避坑指南：我曾经把延迟阈值设成固定值 5 秒，结果业务量翻倍后天天误报。后来改成基于 P95 的动态阈值，世界清净了。记住：阈值要跟着业务量走。
4.3 告警通知策略：让该知道的人知道
告警发出去没人看，等于白设。通知策略的核心就一句话：分级别、分角色、分时段。
我常用的通知矩阵：
| 告警级别 | 通知渠道 | 通知对象 | 通知频率 |
|---|---|---|---|
| Warning | 企业 / Slack 频道 | 值班工程师 | 仅首次，不重复 |
| Critical | 短信 + 电话 | 值班工程师 + 技术负责人 | 每 5 分钟重复一次，最多 3 次 |
| Disaster | 电话 + 全员群 @所有人 | 全体数据团队 + 运维负责人 | 立即通知，持续直到确认 |
这里有个容易被忽略的点：告警去重和聚合。管道出问题时，往往是一连串指标同时报警。如果不做聚合，你会收到几十条告警，根本分不清根因。
我建议用「告警风暴抑制」策略：
- 相同来源的告警，5 分钟内只发一条
- 关联指标（比如延迟和吞吐量同时告警）合并成一条「复合告警」
- 告警恢复后，自动发送一条「已恢复」通知，并关闭工单
一个小技巧：在告警消息里带上「建议排查步骤」和「相关 Dashboard 链接」。这样值班人员不用翻文档，直接点链接就能看。我团队把这个叫「告警自助餐」——收到告警就知道该干嘛。
下面是一个告警通知的伪代码示例，用 Python 写的：
def send_alert(alert_level, metric_name, current_value, threshold):
if alert_level == "Warning":
# 发到工作群，不重复
send_to_channel(metric_name, current_value, threshold)
elif alert_level == "Critical":
# 短信 + 电话，每 5 分钟一次
if not recently_sent(metric_name, 300):
send_sms(oncall_engineer, f"CRITICAL: {metric_name} = {current_value}")
make_phone_call(oncall_engineer)
elif alert_level == "Disaster":
# 电话 + 全员通知
make_phone_call(team_leads)
send_to_all_group(f"🚨 DISASTER: {metric_name} is down!")
create_incident_ticket(metric_name, current_value)4.4 知识体系总览
下面这张图，把监控告警体系的三个核心模块串起来了。你可以把它当成一张「检查清单」——搭建监控时，对着这张图一项项过。
嗯，监控告警这块，说白了就是「采集要全、阈值要准、通知要快」。别想着一步到位，先跑起来，再慢慢调。我团队刚开始的时候，告警准确率只有 60%，迭代了三个月才到 95% 以上。
记住：告警不是越多越好，而是越准越好。一条精准的告警，胜过一百条噪音。