Hive on Spark 企业级调优（二）SQL 优化（上）：执行计划与常规优化

上一篇我们把集群和引擎的"硬件底座"配好了——车道修宽、限速设对。这一篇开始看车怎么开：一条 SQL 进去，Hive 先把它拆成 Stage 和 Operator（第 4 章），然后我们逐个环节抠性能——聚合、裁剪、Join、小文件、并行度、CBO。参数都给你，但更重要的是告诉你每个参数在解决执行计划里的哪一段瓶颈。

4. Hive SQL 执行计划

​使用Explain可以查看 Hive SQL 的执行计划

​Explain 查看的执行计划，有一系列的Stage（这是 Hive 中的 Stage， 跟 Spark 的 Stage 不是一回事）组成，这个 Stage 具有依赖关系，每个 Stage 对应一个 MR Job 或者 Spark Job，或者一个文件系统操作（ load 语句）等。

​每一个 Stage 由一系列的 Operator 组成，一个 Operator 代表一个逻辑草错，例如：TableScan Operator， Select Operator,，Join Operator，Groupby Operator等。

​Stage 与 Operator 的对应关系：

5. 分组聚合优化（map-side）

以这条简单SQL语句为例：

hive>
select
user_id,
count(*)
from dwd_user_login_inc
where dt = '2022-07-01'
group by user_id;

5.1 优化前执行计划

可以使用 Explain 查看执行计划，explain + 执行SQL；

关系如下：

当数据量巨大时，Shuffle 成为最大的性能瓶颈——涉及磁盘 I/O、网络传输、序列化/反序列化。

5.2 优化思路与参数设置

​在这分组聚合的过程中，影响最大的就是 Shuffle 阶段，Shuffle 阶段需要读写磁盘，速度慢。表的 Size 越大，收到的影响越深。所以我们可以开启分组预先聚合，在 Map 1 中就提前先聚合，减少到Shuffle阶段的数据量。

​优化思路为 map-side 聚合。在 map 端维护一个 hash table，利用 hash table 完成部分的聚合，然后将这部分聚合的结果经过 shuffle 发送到 reduce 端，完成最终的聚合。 map-side 聚合能有效减少 shuffle 的数据量，提高 分组聚合 的效率。

​map-side 相关的参数如下：

# 是否在 Map 端进行聚合，默认 true
hive.map.aggr = true;
# hash table 占 map 端内存的大小，默认 0.5
hive.map.aggr.hash.percentmemory = 0.5;
# Map 端聚合后的数据量 / 原始数据量的比值，若大于此值则关闭 Map 端聚合（认为聚合效果不好）
hive.map.aggr.hash.min.reduction = 0.5; 默认 0.5
# 哈希表内存使用达到此值时强制刷写,默认值 0.9
hive.map.aggr.hash.force.flush.memory.threshold = 0.9
# 数据倾斜式是否启用两阶段聚合（详见后续数据倾斜章节），默认 false
hive.groupby.skewindata = false

5.3 优化后的执行计划

​Explain执行计划

示意图：

5.4 执行前后对比

如果基于 Snappy 压缩方式的表，一个分区中的数据落盘大概 4G ，在内存中运算大概 15G：

• 没有优化之前，执行这个 Hive SQL 需要大概 36s 左右
• 开启 map-side 优化之后，执行 Hive SQL 大约在 6s-8s，大大的提高了执行速度

6. 分区裁剪 & 列裁剪

6.1 分区裁剪

— ❌ 全表扫描（扫描所有分区）
SELECT * FROM orders WHERE SUBSTR(order_time, 1, 10) = '2026-07-25';

— ✅ 分区裁剪（只扫描目标分区）
SELECT * FROM orders WHERE dt = '2026-07-25';

— ✅ 多分区范围
SELECT * FROM orders WHERE dt BETWEEN '2026-07-01' AND '2026-07-25';

— ✅ 动态分区裁剪（Spark 3.0+）
SET spark.sql.optimizer.dynamicPartitionPruning.enabled = true;

SELECT o.*, u.user_name
FROM orders o
JOIN users u ON o.user_id = u.user_id
WHERE u.city = '北京'; — 自动裁剪非北京分区

6.2 列裁剪

— ❌ 读取所有列（ORC/Parquet是列存，读所有列=全量I/O）
SELECT * FROM orders WHERE dt = '2026-07-25';

— ✅ 只读需要的列
SELECT order_id, user_id, amount FROM orders WHERE dt = '2026-07-25';

7. 谓词下推(Predicate Pushdown)

​谓词下推：将过滤条件从执行计划的上层移动到下层（更靠近数据源的位置），使得：在数据被读取、传输、计算之前，就尽早地丢弃不需要的数据。

— 开启谓词下推（将WHERE条件下推到存储层过滤）
SET hive.optimize.ppd = true;
SET hive.optimize.ppd.storage = true;

— ORC/Parquet 文件会利用 Row Group 的 min/max 统计信息跳过不相关数据块
SELECT order_id, amount
FROM orders
WHERE dt = '2026-07-25'
AND amount > 1000; — 此条件会下推到文件读取层

8. Join 优化

​Hive 拥有多种 Join 算法，从基础到高级依次为：Common Join → Map Join → Bucket Map Join → Sort Merge Bucket Map Join。优化程度逐级递增，适用条件也逐级严格。

8.1 Common Join

8.1.1 原理

​Common Join 是 Hive 最基础，最稳定的 Join 算法，通过一个完整的 MapReduce / Spark Job 完成

8.1.2 SQL示例

— 订单表 JOIN 用户表（默认走Common Join）
SELECT
o.order_id,
o.amount,
u.user_name,
u.city
FROM orders o
JOIN users u ON o.user_id = u.user_id
WHERE o.dt = '2026-07-25';

8.1.3 执行计划

Stage-1: Map Reduce
Map Operator Tree:
TableScan (orders) → Select → Reduce Output (key: user_id, tag:0)
TableScan (users) → Select → Reduce Output (key: user_id, tag:1)
Reduce Operator Tree:
Join Operator (Inner Join, key: user_id) → Select → File Output

8.1.4 优缺点

8.2 Map Join

8.2.1 原理

​Map Join 将 小表完全加载到分布式内存，在 Map 阶段直接完成 Join，完全跳过 Shuffle 和 Reduce 阶段。

8.2.2 示例SQL

— 方式一：使用 Hint 显式指定 Map Join(不推荐)
SELECT /*+ MAPJOIN(u) */
o.order_id,
o.amount,
u.user_name,
u.city
FROM orders o
JOIN users u ON o.user_id = u.user_id;

— 方式二：开启自动转换（推荐）
SET hive.auto.convert.join = true;
SET hive.mapjoin.smalltable.filesize = 25000000; — 小表阈值25MB

SELECT
o.order_id,
o.amount,
u.user_name,
u.city
FROM orders o
JOIN users u ON o.user_id = u.user_id;

8.2.3 参数设置

# 是否自动将 Common Join 转为 Map Join，默认 true
hive.auto.convert.join = true
# 小表大小阈值，小于此值自动 Map Join，默认 25000000(25MB)
hive.mapjoin.smalltable.filesize = 25000000
# 是否无条件转化（不需要运行时判断），默认 true
hive.auto.convert.join.noconditionaltask = true
# 无条件转化大小阈值，默认 10000000(10MB)
hive.auto.convert.join.noconditionaltask.size = 10000000
# Map Join 后跟 Group By 时哈希表的内存占比，默认 0.3
hive.mapjoin.followby.map.aggr.hash.percentmemory = 0.3

8.2.4 适用条件

• ✅ 大表 Join 小表（小表 < 25MB，可调）
• ✅ 小表能完全装入单个Executor内存，既整个 Hash Map的大小不能超出 14GB
• ❌ 不适用于大表 join 大表
• ❌ 不适用于非等值 Join（如 on a.id > b.id ）

8.2.5 性能对比

Common Join ； 100GB大表 JOIN 20MB小表 -> Shuffle 100GB -> 耗时 30min
Map Join ： 100GB大表 JOIN 20MB小表 -> 无 Shuffle -> 耗时 5min
如果大表的数据量越大，这个差距会越来越大，特别时针对于关联 省份表(数据量小) 时，使用 Map Join 能够极大的提升效率

8.3 Bucket Map Join

8.3.1 原理

​Bucket Map Join 是 Map Join 的扩展，打破了"小表必须完全装入内存"的限制，可用于 大表 JOIN 大表的场景。

​核心思想：如果两张大表都按照 Join Key 进行了分桶（Bucket），那么 桶N 的数据只会与另一张表的 桶M 进行 Join （N于M必须是整数倍关系）。因此 Map 端无需缓存小表全量数据，只需缓存对应桶号的数据。

8.3.2 示例SQL

Step 1 : 创建分桶表

— 订单表：按user_id分8个桶
CREATE TABLE orders_bucketed (
order_id BIGINT,
user_id BIGINT,
amount DECIMAL(10,2),
order_time TIMESTAMP
)
CLUSTERED BY (user_id) INTO 8 BUCKETS
STORED AS ORC;

— 用户表：按user_id分8个桶（或4个桶，必须是倍数关系）
CREATE TABLE users_bucketed (
user_id BIGINT,
user_name STRING,
city STRING,
register_time TIMESTAMP
)
CLUSTERED BY (user_id) INTO 8 BUCKETS
STORED AS ORC;

— 插入数据（必须使用INSERT才能正确分桶）
INSERT OVERWRITE TABLE orders_bucketed
SELECT order_id, user_id, amount, order_time FROM orders;

INSERT OVERWRITE TABLE users_bucketed
SELECT user_id, user_name, city, register_time FROM users;

Step 2 ： 执行 Bucket Map Join

— 设置参数
SET hive.optimize.bucketmapjoin = true;
SET hive.auto.convert.join = true;

— 使用Hint指定
SELECT /*+ MAPJOIN(u) */
o.order_id,
o.amount,
u.user_name,
u.city
FROM orders_bucketed o
JOIN users_bucketed u ON o.user_id = u.user_id;

8.3.3 参数设置

# 是否启用 Bucket Map Join，默认 false
hive.optimize.bucketmapjoin = false
# 需要同时开启 Map Join 自动转化，默认 true
hive.auto.convert.join = true

8.3.4 适用条件（必须满足）

8.4 Sort Merge Bucketed Map Join (SMB Join)

8.4.1 原理

​SMB Join 是 Bucket Map Join 的进一步优化。在 Bucket Map Join 中，每个桶的数据加载到内存之后仍需进行 Hash 匹配。而 SMB Join 要求 桶内数据按 Join Key 排序，这样两个桶的数据就可以进行 归并排序式匹配（Merge Join），无需将任何一个桶完全加载到内存。

8.4.2 示例SQL

Step 1 ： 创建排序分桶表

— 注意：CLUSTERED BY + SORTED BY
CREATE TABLE orders_smb (
order_id BIGINT,
user_id BIGINT,
amount DECIMAL(10,2),
order_time TIMESTAMP
)
CLUSTERED BY (user_id) SORTED BY (user_id) INTO 8 BUCKETS
STORED AS ORC;

CREATE TABLE users_smb (
user_id BIGINT,
user_name STRING,
city STRING,
register_time TIMESTAMP
)
CLUSTERED BY (user_id) SORTED BY (user_id) INTO 8 BUCKETS
STORED AS ORC;

— 插入数据
INSERT OVERWRITE TABLE orders_smb
SELECT order_id, user_id, amount, order_time FROM orders;

INSERT OVERWRITE TABLE users_smb
SELECT user_id, user_name, city, register_time FROM users;

Step 2 ： 执行 SMB Join

— 设置参数
SET hive.optimize.bucketmapjoin = true;
SET hive.optimize.bucketmapjoin.sortedmerge = true;
SET hive.auto.convert.join = true;

SELECT /*+ MAPJOIN(u) */
o.order_id,
o.amount,
u.user_name,
u.city
FROM orders_smb o
JOIN users_smb u ON o.user_id = u.user_id;

8.4.3 参数设置

# 是否启用 SMB Join，默认值 false
hive.optimize.bucketmapjoin.sortedmerge = false
# 必须同时启用 Bucket Map Join 和 Map Join 自动转化
hive.optimize.bucketmapjoin = true
hive.auto.convert.join = true

8.4.4 适用条件（最严格）

8.5 四种 Join 算法对比总结

9. 小文件合并

9.1 小文件产生原因

1. 动态分区写入：每个分区产生独立文件
INSERT INTO TABLE t PARTITION(dt) SELECT …, dt FROM source;
→ 100个分区 × 200个Reducer = 20000个文件！

2. Reduce数量过多：每个Reduce输出一个文件
200个Reducer → 200个文件（可能每个才几MB）

3. 频繁INSERT INTO：每次追加都产生新文件

4. Spark并行度过高：spark.sql.shuffle.partitions=200 → 200个输出文件

9.2 优化方案一：Hive 参数控制合并

— ========== 核心参数 ==========

— 1. 开启Map输出合并（Map-Only任务）
SET hive.merge.mapfiles = true;

— 2. 开启Reduce输出合并（MapReduce任务）
SET hive.merge.mapredfiles = true;

— 3. 【Hive on Spark 专用】开启Spark输出合并
SET hive.merge.sparkfiles = true;

— 4. 合并后目标文件大小（默认256MB）
SET hive.merge.size.per.task = 268435456;

— 5. 触发合并的平均文件大小阈值（小于此值才合并，默认16MB）
SET hive.merge.smallfiles.avgsize = 16000000;

— ========== SQL 示例 ==========
INSERT OVERWRITE TABLE dws_order_daily PARTITION(dt = '2026-07-25')
SELECT
city,
category,
COUNT(*) AS order_cnt,
SUM(amount) AS total_amount
FROM orders
WHERE dt = '2026-07-25'
GROUP BY city, category;
— 执行后，如果输出文件平均大小 < 16MB，会自动启动合并Job

9.3 优化方案二：控制输出文件数量（源头治理）

— 方法1：减少Reduce数量
SET spark.sql.shuffle.partitions = 50; — 从200减到50

— 方法2：使用 DISTRIBUTE BY 控制输出
INSERT OVERWRITE TABLE dws_order_daily PARTITION(dt = '2026-07-25')
SELECT
city,
category,
COUNT(*) AS order_cnt,
SUM(amount) AS total_amount
FROM orders
WHERE dt = '2026-07-25'
GROUP BY city, category
DISTRIBUTE BY city; — 按city分发，相同city的数据写入同一文件

— 方法3：使用 COALESCE Hint（Spark 3.0+）
INSERT OVERWRITE TABLE result_table
SELECT /*+ COALESCE(10) */
city, SUM(amount) AS total
FROM orders
GROUP BY city;
— 强制将输出合并为10个文件

— 方法4：使用 REPARTITION Hint
INSERT OVERWRITE TABLE result_table
SELECT /*+ REPARTITION(10) */
city, SUM(amount) AS total
FROM orders
GROUP BY city;

9.4 优化方案三：输入端合并（CombineHiveInputFormat）

— 每个读表 partition 的最大字节数（默认 128MB，调大=task 变少，调小=task 变多）
SET spark.sql.files.maxPartitionBytes = 268435456; — 256MB
— 打开一个文件的"代价"折算字节，影响小文件是否被合并进同一个 partition
SET spark.sql.files.openCostInBytes = 4194304; — 4MB
— 读表 partition 数下限
SET spark.sql.files.minPartitionNum = 1;

— SQL示例：读取有大量小文件的表
SELECT city, SUM(amount)
FROM orders_with_small_files — 该表有10000个小文件
WHERE dt = '2026-07-25'
GROUP BY city;
— 优化前：10000个Map Task
— 优化后：约 10000×小文件大小 / 256MB ≈ 几十个Map Task

9.5 优化方案四：ORC/Parquet 文件专用合并

— ORC 文件无损合并（不重新计算，仅合并文件）
ALTER TABLE orders PARTITION(dt='2026-07-25') CONCATENATE;

— 或者通过重写实现合并
INSERT OVERWRITE TABLE orders PARTITION(dt='2026-07-25')
SELECT * FROM orders WHERE dt='2026-07-25';

9.6 小文件治理最佳实践

— ========== 完整的ETL任务模板 ==========

— 输入端合并
SET hive.input.format = org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;

— 输出端合并
SET hive.merge.sparkfiles = true;
SET hive.merge.size.per.task = 268435456;
SET hive.merge.smallfiles.avgsize = 64000000;

— 控制并行度
SET spark.sql.shuffle.partitions = 100;

— AQE(后续章节讲解)
SET spark.sql.adaptive.enabled = true;
SET spark.sql.adaptive.coalescePartitions.enabled = true;

— 业务SQL
INSERT OVERWRITE TABLE dws_user_order PARTITION(dt = '2026-07-25')
SELECT
u.user_id,
u.user_name,
COUNT(o.order_id) AS order_cnt,
SUM(o.amount) AS total_amount
FROM users u
LEFT JOIN orders o ON u.user_id = o.user_id AND o.dt = '2026-07-25'
GROUP BY u.user_id, u.user_name;

10. 并行度优化

并行度（Parallelism）决定了任务被拆分成多少个 Task 并行执行：

• 并行度过低： 每个 Task 处理数据量过大，执行慢，集群资源利用不充分
• 并行度过高： Task 数量过多，调度开销大，产生大量小文件，Shuffle 元数据膨胀

10.1 Map 端并行度

Map 端并行度由 输入文件的 Split 数量 决定：

— 控制每个Map处理的数据量
SET hive.input.format = org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;
— 每个Split的最大/最小大小
SET mapreduce.input.fileinputformat.split.maxsize = 256000000; — 256MB
SET mapreduce.input.fileinputformat.split.minsize = 128000000; — 128MB

— 计算公式：
— Map数量 ≈ max(1, min(配置最大数, 总数据量 / split_size))

SQL 示例：

— 场景：100GB数据，默认128MB一个Split → 约800个Map
— 如果集群有200个Core，800个Map需要4轮才能跑完
— 调大Split到512MB → 约200个Map → 1轮跑完

SET mapreduce.input.fileinputformat.split.maxsize = 536870912; — 512MB

SELECT city, SUM(amount)
FROM orders
WHERE dt = '2026-07-25'
GROUP BY city;

10.2 Reduce 端并行度

— Spark引擎下
SET spark.sql.shuffle.partitions = 200; — Shuffle后的分区数（即Reduce并行度）

SQL 示例;

— 场景：50GB数据做Group By
— 默认200个Reducer → 每个处理250MB → 合理
— 如果只有10GB数据 → 每个处理50MB → 并行度过高，调小

SET spark.sql.shuffle.partitions = 50; — 10GB / 50 = 200MB/Task

SELECT
city,
COUNT(*) AS cnt,
SUM(amount) AS total
FROM orders
WHERE dt = '2026-07-25'
GROUP BY city;

10.3 并行度优化经验公式

推荐并行度 = 集群总Core数 × (2~3)

示例：

• 集群：20节点 × 8Core = 160 Core
• 推荐并行度：160 × 2 = 320 ~ 160 × 3 = 480
• 设置：spark.sql.shuffle.partitions = 400
• 每个Task处理数据量建议：128MB ~ 512MB

10.4 完整SQL示例

— 大任务：100GB数据聚合
SET spark.sql.shuffle.partitions = 400;
SET spark.executor.instances = 50;
SET spark.executor.cores = 4;
SET spark.executor.memory = 14g;

SELECT
dt,
city,
category,
COUNT(*) AS order_cnt,
SUM(amount) AS total_amount,
AVG(amount) AS avg_amount
FROM orders
WHERE dt BETWEEN '2026-07-01' AND '2026-07-25'
GROUP BY dt, city, category;

— 小任务：1GB数据查询
SET spark.sql.shuffle.partitions = 20;

SELECT city, COUNT(*) AS cnt
FROM orders
WHERE dt = '2026-07-25'
GROUP BY city;

11. CBO（Cost-Based Optimizer）

11.1 CBO的作用

① JOIN 顺序优化
多表 JOIN 时，决定先 JOIN 哪两张表
→ 让最小的中间结果先产生，避免大表之间先做笛卡尔积

② JOIN 策略选择
根据估算后的表大小，决定用 Broadcast Join 还是 Shuffle Join
→ 小表广播零 Shuffle，大表走 Sort-Merge

③ 聚合位置优化
决定 GROUP BY 放在 JOIN 前还是 JOIN 后
→ 先聚合再 JOIN 可大幅减少 Shuffle 数据量

④ 子查询 / 半连接策略
决定 IN 子查询是物化为临时表做 MapJoin，还是走普通 Shuffle
→ 子查询结果小时物化广播，大时走 Shuffle

11.2 参数设置

— 开启CBO
SET hive.cbo.enable = true;
— 使用统计信息计算查询(必须开启)
SET hive.compute.query.using.stats = true;
— 获取列统计信息
SET hive.stats.fetch.column.stats = true;
— 获取分区统计信息
SET hive.stats.fetch.partition.stats = true;

— 收集表统计信息（CBO依赖）
ANALYZE TABLE orders PARTITION(dt='2026-07-25') COMPUTE STATISTICS;
ANALYZE TABLE orders PARTITION(dt='2026-07-25') COMPUTE STATISTICS FOR COLUMNS;

— CBO会自动：
— 1. 选择最优的Join顺序（多表Join时）
— 2. 选择最优的Join算法
— 3. 选择最优的聚合策略

到这里，一条 SQL 从进入 Hive 到落盘输出，沿途能做的常规优化我们基本走完了：
执行计划怎么看（Explain）→ 聚合怎么提前做（Map-side）→ 数据怎么少读（分区裁剪 / 列裁剪 / 谓词下推）→ 表怎么关联（四种 Join 算法）→ 文件怎么治理（小文件合并）→ 任务怎么切分（并行度）→ 优化器怎么自己选路（CBO）
这些手段有一个共同点：它们都是在 SQL 真正跑起来之前，就把计划定死了。 参数是提前设的，统计信息是提前收集的，Join 顺序是编译期算好的。
但生产环境不会乖乖配合你的统计信息。
下一篇要解决的问题
当"提前规划"失效的时候，怎么办？
举三个你一定遇到过的场景：
场景一：数据倾斜。 你设了 200 个 Reduce，199 个 3 秒跑完，第 200 个跑了 40 分钟还没结束——因为 user_id = -1 的脏数据有 2 亿条，全挤在一个 Task 里。CBO 不管这个，spark.sql.shuffle.partitions 也救不了它。你需要的是从 SQL 层面把倾斜 Key 拆散，或者让引擎运行时自动检测并拆分。
场景二：CBO 猜错了。 统计信息显示 dim_product 有 200 万行，CBO 老老实实选了 Shuffle Join。但实际上一个 WHERE category = ‘electronics’ 过滤完只剩 3MB——本该走 Broadcast Join，白白多了一次全量 Shuffle。编译时做的决策，能不能跑着跑着自己改？ 这就是 AQE（Adaptive Query Execution）要干的事。
场景三：同样的 SQL，你的 Shuffle 比别人慢 5 倍。 不是计划的问题，不是数据量的问题——是序列化。Java 默认序列化带着类名、字段名、继承链一起传，体积膨胀 5~10 倍。换成 Kryo，一个参数的事，Shuffle 时间直接砍到五分之一。再往深了走：Executor 内存怎么分、Storage 和 Execution 各占多少、GC 停顿怎么压到 200ms 以内——这些"最后一公里"的调优，往往决定了任务是从 15 分钟变成 8 分钟，还是从 OOM 变成跑通。
下一篇，我们把这四件事讲透：

最后会把三篇的内容串成一份完整的 ETL 调优模板——从集群配置到 SQL 改写到运行时参数，一个文件搞定，拿来就能贴进生产。