Tez详解

Tez

最近面试pdd发现他们竟然在用 Hive on Tez ，而且很多人对 Tez 的理解，往往停留在“它能提速”这一层。所以写下这篇。

这篇文章就围绕两个问题展开：

1. Tez 是什么

Tez 本质上是一个 DAG 执行引擎。

在 Hive 场景里，SQL 不会直接被“整条执行”，而是先经过编译，转换为物理执行计划，再根据数据依赖和执行边界拆成多个 Vertex。每个 Vertex 可以理解为一个执行阶段，阶段之间通过数据依赖连接起来，形成一张有向无环图（DAG）。

一般来说，当 SQL 中出现以下操作时，往往会引入新的执行阶段：

• Join
• Group By
• Distinct
• Order By
• 需要 Shuffle 的聚合

但要注意：是否真的拆出新的 Vertex，最终还要看优化后的执行计划。

比如某些 Join 可以被优化成 MapJoin，这样就可能减少甚至避免 Shuffle 阶段。

一句话概括：

Tez 的核心，不是把 SQL 一口气跑完，而是把 SQL 拆成多个阶段，让每个阶段以合适的并行度接力执行。

2. Tez 的整体执行过程

一条 Hive SQL 跑到 Tez 上，不是简单地“从上到下执行”，而是会经历几个非常关键的拆分动作：

所以，这条执行链路可以写成下面这样：

Hive SQL
↓
Hive 编译成物理执行计划
↓
识别扫描阶段的输入文件
↓
文件切分成多个 Input Split
↓
多个 Split 经过 Grouping 合并成 Grouped Split
↓
一个 Grouped Split 通常对应一个 Map Task
↓
Map Task 跑在 YARN Container 中
↓
遇到 Join / Group By / Distinct 等操作时发生 Shuffle
↓
上游 Vertex 把数据按 Key 重新分区
↓
下游 Vertex 接收数据继续计算

这样理解会更顺：先有文件切分，再有 Grouping，再有 Task，后面才是 Vertex 之间的 Shuffle 衔接。

更直白一点说，Tez 的执行不是“Vertex 忽然冒出来”，而是先从输入数据开始，把文件切成 Split，再通过 Grouping 决定扫描阶段的 Task 数量；等进入需要重分布数据的阶段后，才会进一步体现出 Vertex 之间的依赖关系。

更直白一点就是：

[SQL]
↓
[执行计划]
↓
[Tez DAG]
↓
[Vertex 1] → [Vertex 2] → [Vertex 3]
↓ ↓ ↓
Task们 Task们 Task们

这里先记住三个概念：

• DAG：任务之间的依赖关系图；
• Vertex：一个执行阶段；
• Task：这个阶段里真正干活的执行单元。

3. 扫描阶段：Input Split 怎么变成 Task

Tez 的并行度，首先从扫描阶段开始看。

3.1 Input Split

Hive 读表时，不会直接把整张表当成一个任务去读，而是先根据 InputFormat 将输入文件切成多个 Input Split。

Split 的大小通常会受到这些因素影响：

• 文件格式；
• 文件大小；
• HDFS Block；
• 压缩方式。

你可以把它理解为：

一大批原始数据，先被切成很多小块，后面再分配给不同的 Task 处理。

3.2 为什么还要 Grouping

如果真的是“一个 Split 对应一个 Task”，那么 Task 数量很容易过多：

• 调度开销大；
• Container 启动开销大；
• 单个 Task 太轻，整体效率反而不高。

Tez 在扫描阶段并不是简单地“一个 Split 对应一个 Map Task”，而是会先做 Grouping。

它的过程可以理解成三步：

举个例子：

假设这一阶段有 96GB 输入数据，当前资源条件下大概支持 8 个 Task 同时跑，split-waves = 2。Tez 先按这个波数把目标并行度定成 16 个 Group，也就是 8 × 2。这样反推出的目标 Group Size 就是 6GB/Group。

如果单个 Input Split 只有 1GB 左右，Tez 就会把 6 个 Split 合并成 1 个 Grouped Split，让每个 Group 的大小尽量接近 6GB；如果 Split 本身已经比较大，或者受到 min-size / max-size 限制，最终分组结果也还是围绕这个目标去调整。

可以把它理解成：

不是先看“一个 Split 能不能单独跑”，而是先看“这一阶段希望开多少个 Task”，再反推每个 Task 应该吃多少数据。

所以，Grouping 的结果不是随机拼出来的，而是围绕“目标并行度 + 目标数据量”来组织的。最终，一个 Grouped Split 通常对应一个 Map Task，因此 Grouped Split 的数量基本就决定了 Map Vertex 的并行度。

3.3 Grouping 的目标是什么

Grouping 其实是在做平衡：

• 太碎了不行，Task 太多；
• 太大了也不行，单个 Task 压力太重。

它想达到的效果是：

每个 Task 都有活干，但不要过重，也不要太轻。

4. Shuffle 阶段：为什么 Join 和 Group By 经常慢

Tez 里另一个最容易慢的地方，就是 Shuffle 阶段。

4.1 Shuffle 是什么

Shuffle 的本质是：

按照 Key，把原本分散在不同 Task 里的数据重新分区，再发给下游 Task。

例如：

上游 Task 1 —> A, B, C
上游 Task 2 —> A, D, E
上游 Task 3 —> B, C, E

洗牌后：

下游 Task 1 —> A
下游 Task 2 —> B
下游 Task 3 —> C
…

你可以把它理解成：

原来大家是乱坐的，现在要按座位号重新排。

4.2 下游并行度怎么定

Shuffle 阶段的并行度，通常不再看 Split，而是更多看 Shuffle 数据量。

Hive 常用这两个参数来估算 Reducer 数：

• hive.exec.reducers.bytes.per.reducer
• hive.exec.reducers.max

它们的作用可以理解为：

这一堆数据，大概要分给几个 Reducer 才合适？

4.3 为什么估算不一定准

现实里经常会遇到这些问题：

• 实际数据量和预估不一致；
• 统计信息不准；
• 某些 Key 分布特别集中；
• SQL 逻辑比较复杂。

所以编译阶段算出来的 Reducer 数，不一定就是最合适的。

4.4 自动并行度的作用

如果开启了 hive.tez.auto.reducer.parallelism，Tez 可以根据运行时实际产生的 Shuffle 数据量动态调整下游并行度。

它的好处是：

• 预估少了，不会让少数 Reducer 扛太多；
• 预估多了，不会开出一堆空转的 Reducer。

所以到了 Shuffle 阶段，真正要关注的不是“Reducer 越多越好”，而是：

每个 Reducer 的数据量是否合理。

5. Tez 调优怎么做

Tez 的调优，本质上就是围绕三个问题展开：

所以，hive.tez.container.size、Grouping 参数和 Reducer 参数，其实都属于同一件事：让每个阶段的 Task 既不要太重，也不要太碎。

5.1 单个 Task 内存不够时，先看 hive.tez.container.size

hive.tez.container.size 主要控制的是 Tez Task 向 YARN 申请的 Container 内存大小。

它影响的是单个 Task 的资源规格，不是直接决定 Task 数量。

如果你看到这些现象：

• Join 很吃内存；
• 聚合很吃内存；
• 排序很吃内存；
• 频繁 OOM；
• Full GC 很多。

那么可以考虑把 Container 调大一些。

但如果问题本质是并行度不够，就不能只加内存。

5.2 扫描阶段并行度不够时，先动 Grouping

如果问题本质是：

• Task 太少；
• 并行度不足；
• 单个 Task 数据量太大。

那优先不要先加 hive.tez.container.size，而是先动这些参数：

• tez.grouping.split-waves
• tez.grouping.min-size
• tez.grouping.max-size

这几个参数的目的很明确：控制输入 Split 怎么合并，进而决定 Map Task 的数量。

如果扫描阶段 Task 太少：

• 提高 tez.grouping.split-waves；
• 适当调小 tez.grouping.min-size；
• 必要时检查 tez.grouping.max-size 是否把 Group 卡得太大。

这样做的目的，是让更多 Split 被分开处理，提升 Map 并行度。

5.3 Shuffle 阶段并行度不够时，先动 Reducer 相关参数

如果问题出在 Join、Group By 这类需要 Shuffle 的阶段，那重点就不是 Grouping，而是 Reducer 并行度。

这时优先看：

• hive.exec.reducers.bytes.per.reducer
• hive.exec.reducers.max
• hive.tez.auto.reducer.parallelism

如果 Shuffle 阶段并行度不够：

• 把 hive.exec.reducers.bytes.per.reducer 调小；
• 必要时提高 hive.exec.reducers.max；
• 开启 hive.tez.auto.reducer.parallelism，让运行时根据实际 Shuffle 数据量动态调整。

这样做的目的，是让每个 Reducer 分到的数据更合理，避免少数 Reducer 扛太多数据，也避免下游并行度太低。

5.5 怎么判断该动哪一类参数

你可以直接按这个顺序判断：

• 先看是不是单个 Task 内存不够，如果是，先动 hive.tez.container.size；
• 再看是不是扫描阶段并行度不够，如果是，先动 Grouping 参数；
• 再看是不是 Shuffle 阶段并行度不够，如果是，先动 Reducer 相关参数；
• 如果少数 Task 特别慢，再去看数据倾斜、超大文件和数据本地性。

一句话：

调优不是把参数分开看，而是先判断问题出在哪个阶段，再去动对应的参数。

7. 总结

Tez 会把 Hive SQL 拆成多个阶段，每个阶段再拆成多个 Task；扫描阶段主要通过 Grouping 决定 Map 并行度，Shuffle 阶段主要通过 Shuffle 数据量和 Reducer 估算决定下游并行度。