倾斜爆发才来处理已经太晚了——怎么建立数据倾斜的监控和预警体系？

-- 数据分布监控SQL（每天跑一次，写入监控表）
INSERT INTO data_distribution_monitor
SELECT
  '2024-08-01' AS dt,
  'orders' AS table_name,
  'user_id' AS key_col,
  user_id AS key_value,
  cnt,
  cnt * 100.0 / total_cnt AS pct,  -- 占总量百分比
  RANK() OVER(ORDER BY cnt DESC) AS rnk  -- 排名
FROM (
  SELECT
    user_id,
    COUNT(*) AS cnt,
    SUM(COUNT(*)) OVER() AS total_cnt
  FROM orders
  WHERE dt = '2024-08-01'
  GROUP BY user_id
) t;
 
-- 预警查询：TOP 10 key累计占比超过30%就预警
SELECT dt, SUM(pct) AS top10_pct
FROM data_distribution_monitor
WHERE table_name = 'orders' AND key_col = 'user_id' AND rnk <= 10
GROUP BY dt
HAVING SUM(pct) > 30  -- 超过30%触发预警
ORDER BY dt;

# 通过Spark REST API获取Stage的Task分布指标
import requests
import json
 
SPARK_UI_URL = 'http://spark-master:4040'  # 替换为实际地址
 
def check_skew(app_id, stage_id, threshold=5.0):
    # 获取Stage的所有Task信息
    url = f'{SPARK_UI_URL}/api/v1/applications/{app_id}/stages/{stage_id}'
    resp = requests.get(url)
    stage_data = resp.json()
 
    # 获取Task列表
    tasks_url = f'{SPARK_UI_URL}/api/v1/applications/{app_id}/stages/{stage_id}/taskList'
    tasks = requests.get(tasks_url).json()
 
    durations = [t['taskMetrics']['executorRunTime'] for t in tasks
                 if 'taskMetrics' in t]
    if not durations:
        return None
 
    max_dur = max(durations)
    avg_dur = sum(durations) / len(durations)
    ratio = max_dur / avg_dur if avg_dur > 0 else 0
 
    # 最慢Task超过平均Task的5倍 = 倾斜预警
    if ratio > threshold:
        return {
            'stage_id': stage_id,
            'max_duration_ms': max_dur,
            'avg_duration_ms': avg_dur,
            'ratio': ratio,
            'is_skewed': True
        }
    return None

# Flink指标通过REST API或Prometheus获取
# Flink暴露的关键指标：
 
# 每个SubTask的numRecordsInPerSecond（每秒接收记录数）
# 对比各SubTask的这个指标，差异超过5倍 = 倾斜预警
 
def check_flink_skew(job_id, operator_id, threshold=5.0):
    url = f'http://flink-master:8081/jobs/{job_id}/vertices/{operator_id}/subtasks'
    resp = requests.get(url).json()
    subtasks = resp.get('subtasks', [])
 
    # 提取每个SubTask的接收记录数
    records_in = []
    for st in subtasks:
        metrics = st.get('metrics', {})
        records_in.append(metrics.get('numRecordsIn', 0))
 
    if not records_in or max(records_in) == 0:
        return None
 
    ratio = max(records_in) / (sum(records_in) / len(records_in))
    if ratio > threshold:
        return {
            'operator': operator_id,
            'max_records': max(records_in),
            'avg_records': sum(records_in) / len(records_in),
            'skew_ratio': ratio
        }
    return None

建立完整的倾斜预警体系

光有监控脚本还不够，要把它们集成到数据平台的监控体系里：

层次	监控内容	告警条件	处理方式
数据分布层	TOP N key占总量的百分比	TOP10占比 > 30%，且较上周上升 > 5%	人工审查，考虑优化
作业运行层	Spark Stage最慢Task/平均Task时间比	比值 > 5，且Stage持续超过30分钟	自动告警，人工处理
Flink流处理层	各SubTask的numRecordsIn差异	最大/平均 > 5	立即告警，实时作业影响大
SLA层	作业完成时间vs预期SLA	超SLA 20%以上	升级告警，影响业务

在实际落地时，监控告警通常接入公司的统一告警平台（Alertmanager + PagerDuty，或者简单的企业微信/钉钉机器人告警）。关键是告警要有「上下文」：不只说「作业慢了」，要说「orders表的user_id字段，TOP 10 key今天占比34%（上周26%），建议检查是否有新的异常账号产生大量数据」——有上下文的告警，处理效率能提高5倍以上。

数据画像：提前识别高风险作业

另一个维度是「数据画像」：定期给每个作业建立性能画像，记录历史的Task分布特征，自动识别「高倾斜风险」的作业，在倾斜爆发前就安排优化。

-- 作业画像表：记录每次作业运行的倾斜指标
CREATE TABLE job_performance_profile (
  job_name STRING,
  run_date DATE,
  stage_id INT,
  max_task_duration_sec BIGINT,
  avg_task_duration_sec BIGINT,
  skew_ratio DOUBLE,  -- max/avg，越大说明倾斜越严重
  total_shuffle_bytes BIGINT,
  max_task_shuffle_bytes BIGINT,
  shuffle_skew_ratio DOUBLE  -- Shuffle数据量倾斜比
);
 
-- 查询高风险作业：skew_ratio连续3天上升
SELECT
  job_name,
  AVG(skew_ratio) AS avg_skew_recent_7d,
  MAX(skew_ratio) AS max_skew_recent_7d,
  -- 计算trend：最近7天的skew_ratio线性趋势
  (MAX(CASE WHEN run_date = CURRENT_DATE THEN skew_ratio END) -
   MAX(CASE WHEN run_date = CURRENT_DATE - 7 THEN skew_ratio END)) AS skew_growth_7d
FROM job_performance_profile
WHERE run_date BETWEEN CURRENT_DATE - 7 AND CURRENT_DATE
GROUP BY job_name
HAVING avg_skew_recent_7d > 3  -- 近7天平均倾斜比超过3
   AND skew_growth_7d > 1      -- 且7天内倾斜比上升超过1
ORDER BY skew_growth_7d DESC;
-- 这些作业是「高风险」，需要提前介入优化

面试高频追问

追问一：监控数据分布的全表扫描代价太大，生产环境怎么解决？

有几个低成本的替代方案：1）采样统计，TABLESAMPLE(0.1 PERCENT)只扫描0.1%的数据，速度提升1000倍，对分布估算的误差通常在可接受范围（±5%）；2）在数据写入时埋点，比如Flink写入Kafka时，顺便统计每个key的count并写到一个轻量的统计表，代价极低；3）利用Parquet/ORC文件的列统计信息（min/max/count存在文件metadata里），通过读metadata而不是读数据来估算分布，Spark的spark.read.parquet可以直接读metadata；4）对于固定的高风险字段（如user_id、device_id），建立专门的基数监控指标，每小时更新一次，不需要全表扫描。

追问二：倾斜监控的告警阈值怎么定？5倍是怎么来的？

5倍是经验值，不是科学定论。定阈值的原则：1）宁可多报（降低阈值）也不要漏报，因为倾斜爆发的代价比误报高得多；2）阈值应该根据业务的SLA要求来定，如果作业SLA是1小时，倾斜比3倍就可能导致SLA超时，阈值应该设2-3倍；3）不同类型的作业阈值不同，实时作业的阈值要比批处理作业低（实时倾斜危害更大）；4）初期设宽一点（10倍），观察告警频率，根据实际运营经验逐步收窄。我们团队最终落地的阈值是：批处理5倍告警+10倍严重告警，Flink实时3倍告警+5倍严重告警。

追问三：除了倾斜，数据质量问题（脏数据、NULL暴增）会不会被这套监控体系捕捉到？

会的，而且这套监控体系的价值不只是倾斜，还能捕捉数据质量异常：1）NULL暴增——如果NULL key占比从1%暴涨到20%，会触发「TOP key占比异常」的告警，因为NULL是一个超大key；2）数据量异常——如果某天的总行数比历史均值高出50%，说明可能有重复数据写入或上游数据异常，这是另一个监控维度；3）新热点key出现——新的高频key突然出现（比如新的营销活动带来的爆款商品），会在数据分布监控里体现。所以这套「数据分布监控+作业性能画像」的体系，既是倾斜预警，也是数据质量监控的一部分。

实战踩坑

我在某短剧出海公司建立了一套轻量级的倾斜监控体系。当时我们的广告归因作业每天要处理跨Facebook、TikTok、Google三个平台的广告数据，某次一个爆款广告的曝光量在3天内从日均50万次涨到5000万次，导致按ad_id GROUP BY的作业从原来的20分钟变成了4个小时，影响了次日的报表准时产出。

-- 我们当时建立的简版分布监控（每日运行，写入监控表）
INSERT OVERWRITE TABLE ad_distribution_monitor PARTITION(dt='2024-08-01')
SELECT
  'ad_id' AS dim_name,
  ad_id AS dim_value,
  COUNT(*) AS event_cnt,
  COUNT(*) * 100.0 / SUM(COUNT(*)) OVER() AS pct
FROM ad_exposure_dwd
WHERE dt = '2024-08-01'
GROUP BY ad_id
ORDER BY event_cnt DESC
LIMIT 100;  -- 只存TOP 100，代价低
 
-- 预警逻辑（在DataWorks调度里配置，日报告警）
-- 如果当天TOP 1 ad_id的pct > 昨天TOP 1 ad_id的pct * 2，发告警
SELECT
  t1.dt AS today,
  t1.dim_value AS top_ad_id,
  t1.pct AS today_pct,
  t0.pct AS yesterday_pct,
  t1.pct / t0.pct AS growth_ratio
FROM ad_distribution_monitor t1
JOIN ad_distribution_monitor t0
  ON t1.dim_value = t0.dim_value
  AND t1.dt = '2024-08-01'
  AND t0.dt = '2024-07-31'
WHERE t1.pct = (
  SELECT MAX(pct) FROM ad_distribution_monitor WHERE dt='2024-08-01'
)
HAVING growth_ratio > 2;  -- TOP 1 key占比翻倍就告警

效果：这套监控建立后，有2次成功提前发现了倾斜风险：一次是爆款广告占比从8%涨到19%，告警触发后我们提前加了加盐处理，作业正常运行；另一次是某个测试广告ID数据写入错误，占比暴涨到60%，告警触发后快速排查到是上游数据管道配置错误，及时修复，没有影响业务报表。

总结

「发现」倾斜和「处理」倾斜一样重要。建立两层监控：数据分布层（TOP N key占比趋势）和作业运行层（Task时间分布极端值）。告警要有上下文，不只说慢了，要说是哪个key、占比多少、比上周涨了多少。结合作业性能画像，提前识别高风险作业，在倾斜爆发前就介入优化。面试时能讲出「监控体系」的设计，而不只是「遇到倾斜怎么处理」，会让面试官觉得你有工程体系化思维，而不只是能处理一次次的「救火」。

数据倾斜系列已经讲完了10篇核心内容！下一篇开始讲「面试表达」——遇到倾斜相关的面试题，怎么组织语言、用STAR法则讲出最高分的答案？

你们有没有建立倾斜相关的监控体系？是怎么落地的？评论区分享你们的方案～

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

数据倾斜

面试通关手册

第11-15篇：OLAP倾斜 · 小文件 · 面试表达 · 多表JOIN · 故意倾斜

Wood聊大数据

9年大数据老兵，聊技术、聊职场、聊成长

面试杀手题🔥🔥 | 高阶篇 | 每篇附实战案例