🚨 数据倾斜调优面试题

数据倾斜是分布式计算（Hive/Spark）的 “性能杀手”—— 指 shuffle 阶段（group by/join/distinct等操作）中，部分 Task 处理的数据量是其他 Task 的几十倍甚至几百倍，导致节点负载不均、任务卡断 / 超时 / OOM 失败。以下是详细到 “原理 + 场景 + 实操” 的调优方案：

⚙️一、参数调优：用配置 “驯服” 倾斜

参数调优是解决倾斜的 “基础操作”，针对 Hive/Spark 分别适配核心配置：

（一）Hive 参数调优

通过会话参数改写执行逻辑，从根源减少倾斜风险：

✅ set hive.auto.convert.join=true;
- 作用：强制启用自动 Map Join，将 “小表 + 大表” 的 Join 从 Reduce 阶段提前到 Map 阶段完成。
- 原理：任务启动时，把小表全量加载到每个 MapTask 的内存（以 Hash 表形式存储）；大表的 MapTask 读取数据时，直接在本地与内存中的小表匹配，完全跳过 Reduce 的 shuffle。
- 适用场景：小表（如维度表）关联大表（如事实表），比如 “用户表（100M）关联订单表（100G）”。
- 注意：若小表超内存会 OOM，需配合下一个参数设阈值。
📏 set hive.mapjoin.smalltable.filesize=100M;（思维导图 “100t” 为笔误）
- 作用：定义 “小表” 的阈值 —— 表大小≤该值时，自动触发 Map Join；超过则回退为 Reduce Join。
- 实操建议：根据集群内存调整（比如 Yarn 单容器内存 8G 时，可设为 2G）。
🧩 set hive.groupby.skewindata=true;
- 作用：解决group by倾斜，自动拆分为2 个 Job 分步计算：
  - Job1：给所有数据打随机后缀，分散到多个 Reduce 做 “预聚合”（比如把大 Key 拆成多个子 Key）；
  - Job2：去掉随机后缀，对预聚合结果做 “最终聚合”。
- 适用场景：某几个 Key 占比超 50% 的group by（比如电商 “爆款商品” 的销量统计）。
📦 set hive.merge.mapfiles=true;
- 作用：Map 阶段结束后，合并小文件（仅适用于 Map-only 任务）。
- 原理：小文件过多会导致后续 Reduce 阶段的 Task 数激增，合并后减少资源开销。
🧠 set mapreduce.map.memory.mb=4096;/set mapreduce.reduce.memory.mb=4096;
- 作用：分别给 Map/Reduce Task 分配内存（单位：MB）。
- 适用场景：数据倾斜时，单个 Task 处理数据量过大，默认内存不足导致 OOM。

（二）Spark 参数调优

Spark 的 “自适应执行（AQE）” 是倾斜调优的 “利器”，核心配置如下：

🚀 spark.sql.adaptive.enabled=true;
- 作用：启用自适应查询执行—— 任务运行时动态调整执行计划（不用提前写死配置）。
- 核心能力：
  - 动态调整 shuffle 分区数；
  - 小表自动转广播 Join；
  - 合并小 Task 减少资源浪费。
🎛️ spark.sql.shuffle.partitions=200;（默认值）
- 作用：设置 shuffle 的初始分区数；开启 AQE 后，会根据数据量动态增 / 减分区（比如数据少则合并，数据多则拆分）。
- 实操建议：默认 200 在大数据量时可能太少，可先设为 500。
📊 spark.sql.adaptive.advisoryPartitionSizeInBytes=256M;（思维导图 “2048M” 为笔误）
- 作用：AQE 的 “建议分区大小”—— 自适应会把分区调整到接近该值（比如数据量 100G 时，会拆成约 400 个分区）。
- 常用值：128M~512M（根据集群单 Task 处理能力调整）。
📡 spark.sql.autoBroadcastJoinThreshold=10M;
- 作用：设置 “广播小表” 的阈值 —— 表≤该值时，自动转为 Map Join（广播到所有 Executor 内存）。
- 注意：若小表超阈值，会回退为 Shuffle Join，需手动拆分大表。

🗺️二、Map 阶段调优：从源头 “削平” 倾斜

Map 阶段是数据处理的 “入口”，优化输入可大幅减少后续倾斜风险：

✂️ 剪枝列 & 剪条件：减少数据量
- 剪枝列：只查需要的字段，避免select *（比如统计订单金额，只查order_id和amount，不查用户地址 / 备注）。
  - 效果：减少 Map 端读取的数据量，间接降低后续 shuffle 的压力。
- 剪条件：提前过滤无效数据（比如空值、测试数据、过期数据）。
  - 示例：where status != 'test' and create_time >= '2023-01-01'，过滤后数据量可能减少 30%+。
🎲 distribute by rand()：随机打散数据
- 作用：让 Map 端输出随机分配到 Reduce，避免某类 Key 集中到单个 Reduce。
- 原理：原本 “Key=A” 的 100 万条数据会全发往 Reduce1，加distribute by rand()后，会分散到所有 Reduce。
- 适用场景：数据分布极不均（比如某 Key 占比超 60%）的group by/join，比如 “热门主播的弹幕统计”。
- 示例：
  
  sql
```
select user_id, count(*) 
from danmu_table 
distribute by rand()  -- 随机打散
group by user_id;
```
🚫 用group by替代distinct
- 问题：select distinct user_id from table会把所有数据发往单个 Reduce（因为要全局去重），极易倾斜。
- 优化：select user_id from table group by user_id—— 数据会分布式分配到多个 Reduce 处理，避免单点压力。

🔧三、Reduce 阶段调优：解决 shuffle 核心倾斜

Reduce 阶段是倾斜的 “重灾区”，针对不同场景精准突破：

（一）Key 类倾斜优化

📉 小 Key 优化：先过滤再关联

场景：大量低频小 Key（出现次数 < 10 次）占用 Reduce 资源，导致空跑。
优化逻辑：先筛选 “高频 Key”，再关联其他表（小 Key 可后续单独处理）。

示例：

sql

-- 步骤1：筛选高频Key（出现次数>10）
create temporary table high_freq_keys as
select user_id from log_table group by user_id having count(*) > 10;

-- 步骤2：仅用高频Key关联其他表
select a.user_id, b.order_id
from log_table a
join order_table b on a.user_id = b.user_id
where a.user_id in (select user_id from high_freq_keys);

📦 大 Key 优化：先拆分再聚合

场景：单个大 Key（比如某用户 ID 对应 100 万条数据）导致单个 Reduce 负载过高。
优化逻辑：给大 Key 加随机后缀拆分→ 分散聚合→ 合并结果。

实操步骤：

sql

-- 步骤1：给大Key加随机后缀（比如拆成5份）
create temporary table split_big_key as
select 
  if(user_id = 'big_key_123', concat(user_id, '_', floor(rand()*5)), user_id) as new_user_id,
  amount
from order_table;

-- 步骤2：按新Key预聚合
create temporary table pre_agg as
select new_user_id, sum(amount) as sum_amount
from split_big_key
group by new_user_id;

-- 步骤3：去掉后缀，最终聚合
select 
  split(new_user_id, '_')[0] as user_id,
  sum(sum_amount) as total_amount
from pre_agg
group by split(new_user_id, '_')[0];

🔄 大 Key 重计算：打随机值分散
- 场景：大 Key 无法拆分但数据量极大（比如 “全平台总销量” 的统计）。
- 优化：给大 Key 打随机值（比如key_1~key_10），分散到多个 Reduce 计算后，再合并结果。

（二）Map Join 优化：完全跳过 shuffle

原理：将小表全量加载到 Executor 内存，在 Map 端直接与大表数据匹配，完全避免 Reduce 阶段的 shuffle。
适用场景：

① 小表 + 大表的 Join（比如 “商品维度表（50M）关联交易表（200G）”）；

② 大表 Join 大表，但其中一方的 Key 集中在小范围（比如 “订单表 Join 用户表，仅关联新注册用户”）。
Spark 开启方式：set spark.sql.autoBroadcastJoinThreshold=50M;（调整小表阈值）。

📦四、其他调优：解决 “间接倾斜” 问题

小文件、存储布局会间接引发倾斜，需配套优化：

（一）小文件优化：减少 Task 激增

小文件过多会导致 MapTask 数爆炸（比如 1 万个小文件→1 万个 MapTask），引发资源浪费：

🗂️ 分区分桶策略：
- 用Hash 分桶替代普通分区：比如create table order_table clustered by (user_id) into 10 buckets，让数据均匀分布到 10 个桶中，避免某分区数据量过大。
🧹 小文件治理：
- Spark 开启 AQE 自动合并小文件；
- Hive 设置hive.merge.mapredfiles=true合并 Reduce 小文件；
- 定时运行 “小文件合并 Job”（比如每天凌晨合并前一天的日志表）。

（二）Z-Order 排序：优化数据存储布局

作用：多维度排序（比如ZORDER BY user_id, create_time），让 “相关数据物理上存储在一起”。
原理：原本无序的数据中，同一用户的不同时间数据可能分散在不同文件；Z-Order 后，同一用户的所有数据会集中存储，查询时减少 IO 扫描范围。
适用场景：多维分析表（比如按 “用户 + 时间” 查询的交易表）。

（三）Spark3.0 AQE：“智能调优” 解放双手

核心能力：任务运行时动态调整执行计划，无需手动写复杂配置：

① 动态拆分大分区（解决倾斜）；

② 小表自动转广播 Join；

③ 合并小 Task 减少资源浪费。
适用场景：复杂查询、数据分布不确定的任务，大幅降低调试成本。