Apache Hive 是怎样做基于代价的优化的？

Lenis

发布于 2019-12-25 16:15:34

1K0

发布于 2019-12-25 16:15:34

文章被收录于专栏：有关SQL有关SQL

上一篇文章 Apache Calcite 为什么能这么流行末尾提到要单独开一篇文章，聊下 Hive 怎么利用 Calcite 做基于代价查询优化，现在兑现承诺。

基于代价的优化器

通常，我们把 SQL 查询优化器分为两种类型：

RBO(Rule Based Optimizer)
CBO(Cost Based Optimizer)

RBO 顾名思义，就是事先定义好一系列的规则，然后去遍历这些规则做优化。

而 CBO，自然就是根据所谓的代价去做优化，代价最小的执行计划就是最好的执行计划。

RBO 固然是好的，能解决很多问题。

这是上一篇文章里的例子，一个很简单的查询，对应的执行计划是这样：

通过两个常见的规则转换，就能得到下面这个更好的执行计划：

RBO 好不好，很好嘛，project 和 filter 都 push down 之后不就能大大减小数据量了，性能不就好了嘛。

但是 RBO 还不够好：

规则是基于经验的，经验就可能是有偏的，总有些问题经验解决不了
不太可能列出所有经验，事实上这些规则也确实是逐渐充实的

Hive 里的 CBO

Hive 在 0.14 版本引入了 CBO，典型的，由于 join 是 SQL 中非常影响性能的操作，所以引入之初就解决了下面几个大难题：

Join Ordering Optimization
Bushy Join Support
Join Simplification

很显然，我们光看名字就知道，这几个问题不是 RBO 能解决了。篇幅有限，我们只看第一类情况。

这个例子来自 TPC-DS Q3，比刚才那个例子稍微复杂一点。但也就是多了一张表一起 join，再多一些过滤条件。

很显然，这个查询依然能受益于 RBO 里的 push down 规则。另外留意下，两个表过滤之后的行数是这样：

下面对比下，RBO 之后的执行计划是这样：

而经过 CBO 之后的执行计划是这样的：

可以看到，store_sales join item 之后的结果只有 82 million 行，比默认的 store_sales join date_dim 的 14 billion 行少了一个数量级了。

不同的 join 顺序带来的性能差距是巨大的。实际的性能测试结果会更直观：

很显然，RBO 是没法做到这点的。没法总结出这么条规则，来判断哪个表应该放在 join 顺序的前面。

那 CBO 又是怎么做到的呢？

定义代价模型

不难看出，上面的例子中，主要是通过这么两点来判断 join 顺序的：

原始表的行数
过滤之后的行数

说白了，就是行要少，无论是原始数据的行，还是中间结果的行，越少性能越好。

那是不是就用行来衡量代价就够了呢？

没这么简单，因为影响性能的不只有行。

比如

更小的数据体积
更高的并发度（前面提到的 Bushy Join 优化就有涉及）

也是能大幅提高性能的，而这都不是行数能体现的。

退一步看，行数作为代价不够理想，一方面是因为不够直接，所以表达力有限；另一方面，是因为看起来又像走回了规则的老路。

我们需要一个更好的代价模型。

试想一下，代价的本质是什么？是对资源的消耗。

一个计算机系统，最基本的资源是什么？

CPU
Memory
IO
- Disk IO
- Network IO

直接把代价对应到资源的消耗不就完了吗，搞定。

还不够。

Hive 的数据是存在 HDFS 上的，所有对 HDFS 上的数据的读写都得经过 HDFS，而不能直接操作磁盘。所以有一部分的 IO 实际上是走的 HDFS，并且由于数据本地性的存在，没法知道这部分 IO 是 Disk IO 还是 Network IO。因此需要把 HDFS IO 单列出来。

而内存，可能由于在计算过程中是动态使用的，由于实际的操作和算法的不同，很难去准确计算，同时各种计算框架往往在内存不够用的情况下会 spill 到磁盘，反过来干扰 Disk IO 的计算。类似的原因使得几乎所有存储引擎和计算引擎在计算代价的时候都没有把内存考虑在内。

所以，我们得到这么一个代价模型，更准确点，代价参数：

CPU
IO
- HDFS IO
- Disk IO
- Network IO

计算代价

那怎么把实际 SQL 的消耗计算成 CPU 和 IO 的消耗呢？

Hive 定义了上图这些代价变量，我用不同的颜色来标识分组。

黄色代表 HDFS IO，灰色代表 Disk IO，橙色代表 Network IO，紫色代表数据属性，红色代表 CPU。

来看几个典型的例子。

Table Scan Cost

CPU Cost = 0
IO Cost = Hr * T(R) * Tsz

很好理解，表的扫描完全是 HDFS IO 操作。

Map Join Cost

CPU Cost = HashTable Construction cost + Cost of Join = ((T(R2) + …+ T(Rm)) + (T(R1) + T(R2) + …+ T(Rm))) * CPUc nano seconds
IO Cost = Cost of transferring small tables to Join Operator Node * Parallelization of the join = NEt * (T(R2) * Tsz2 + … + T(Rm) * Tszm) * number of mappers
Number of Rows = Join Cardinality Estimation

稍微复杂点，思考下 Map Join 的原理，不难知道 CPU 的消耗由小表 HashTable 的创建和各表 join 的消耗组成。而 IO 的消耗则是把各个小表广播到大表对应的 mapper 上去的 Network IO 开销。有个之前没出现的东西， mapper 的数量，但这个值是可以根据文件格式、大小来确定的，这是由 MapReduce 的原理决定的。

Filter Cost

CPU Cost = T(R) * CPUc nano seconds
IO Cost = 0
Number of Rows = Filter Selectivity * Number of Rows from Child

过滤则是典型的纯 CPU 操作。注意过滤的时候，实际已经拿到数据了， IO 开销在之前的 Table Scan 操作就付过了。

这里又出现了一个上面代价变量里没有的东西 -- Selectivity。前面那个 TPC-DS 的例子里，我们知道了这个东西代表数据过滤完剩下的比例，越小越好。但这个值却不像刚才的 mapper 数量那么好算。

考虑上图这种情况，我们知道了 c_id 这列的最大、最小值，也知道了 distinct 值，怎么去算 c_id > N 的数量呢？

c_id.distinct(after_filter) = 
(c_id.Max – N) / (c_id.Max – c_id.Min) * c_id.distinct(before_filter)

这个算法很简单直接，但很显然是有前提的。前提就是，数据的分布必须是均匀的。

但更显然的是，数据的分布通常都是不均匀的，通常更好的做法是有个所谓的 histogram，也就是直方图，来表示数据的真实分布。

Hive 提供了 histogram_numeric 函数来以直方图的形式计算数据的分布，会起一个 MR 任务去做计算。但可惜的是数据并不会写入 metadata，也就无法作为下次查询的优化依据。

类似上面的三个例子，我们可以把所有操作的代价计算方法都定义清楚，这样每一步操作的代价就都明确了。

最后，怎么计算一个执行计划最终的代价呢？

我们知道，查询引擎是以一个树（Operator Tree）的形式去构造和优化查询计划的，而每个节点都是实际需要执行的操作（Operator）。我们能计算每个节点的代价，那把所有节点的代价累加起来，就是整个执行计划的代价。

再看一眼刚才这张图，上面我们也提到，紫色代表数据属性。既然是数据属性，那就和实际数据直接相关，那怎么拿到这些数据呢？

通过 Analyze 命令获取数据属性

执行 desc 命令可以看到类似上图的 Partition Parameters。

desc formatted [table_name] partition([partition_expr])

可以看到，numFiles、totalSize 和 transient_lastDdlTime 都是有值的，但 numRows 和 rawDataSize 的值却是 -1，也就是值不确定。

执行一下这个命令：

analyze table [table_name] partition([partition_expr]) compute statistics;

发现会启动一个 MR 任务，执行完之后再 desc 这个分区：

numRows 和 rawDataSize 都有了值。

通过 Analyze 命令，我们就能得到数据的属性，并且这些属性是持久化到 metastore 的，以后所有对这个表的查询都可以用这些数据来做优化。

这还只是对表和分区的统计结果，对于像 Filter 这样的操作，显然是不够的。很简单，只要在刚才的命令后面加上 for columns 就可以对列做相关的数据分析。

analyze table [table_name] partition([partition_expr]) compute statistics for columns;

同样是会起一个 MR 任务，通过 desc 命令

describe formatted [table_name] [column_name] partition(data_date=20180523);

就能看到对应列的统计分析结果：

视数据类型的不同，会在不同的列算出对应的分析结果。

看完上面这一段，很自然会有些问题，比如为什么表的统计分析有些数据有结果，有些又没有？既然要起 MR 任务，那肯定会很消耗资源，并且可能影响线上任务咯？

是的没错，答案可以简单的概括如下：

hive.stats.autogather=true
- 默认为 true，在用 create、insert 命令创建表时会自动生成统计数据，毕竟都是 MR 嘛，顺便就算了
- 对于类似 load 或者添加文件到外表分区这样的操作，就不会自动更新统计数据了，毕竟没有 MR 任务嘛
hive.stats.column.autogather=false
- 默认为 false，只能手动触发；或者如果能接受，把这个值改为 true 变成自动触发，自动触发条件和上面类似
Analyze 命令会启动 MR 任务，毕竟消耗资源，所以要明智的使用
- 只对发生变化的数据使用，只在发生变化之后使用
- 只对频繁使用的数据（可以只是部分列）使用
- 在系统不忙的时候使用

说了这么多，还是没和上一篇搭上线，到底 Hive 的 CBO 和 Calcite 有什么关系呢？

Hive 是怎么利用 Calcite 做的 CBO

Hive 在 0.14 版本终于引入了 CBO，这个在传统关系数据库里几乎是标配的东西。

早期的包结构和依赖的项目名是这样：

一直演进到现在变成这样：

总算看到了 Calcite。

回过头来，再看看 Calcite 的架构图：

上一篇也提到，正是由于 Calcite 灵活可插拔的架构，使得 Hive 可以完全使用自己独立的 SQL Parser 和 Validator，而只用 Calcite 的 Query Optimizer。

而 Hive 在代码层面和 Calcite 的结合体现在 CalcitePlanner 这个类：

这个类里又重点关注两个地方，一个就是上图选中的 HiveVolcanoPlanner。

这就是 HiveVolcanoPlanner 完整的代码，是的，就这么一点。可以看到除了传入 HiveCost.FACTORY 作为参数初始化对象外，其他都直接用父类的。

从名字不难猜出，Hive 是想定义自己的 cost funciton，不想用 Calcite 默认的 cost function。

是有多差啊，还不想用。看一眼，长这样：

确实，简单粗暴，谁行少谁更优。

那 HiveCost 改成啥样了呢？

咋还是只看行呢？注意看下面的注释。上图是早期的版本，在比较新的版本，注释已经被扶正了。

意思也很容易理解，在 Hive 看来，CPU 和 IO 应该优先级比行数更高，先比较这俩，如果相等，才去看行数。而 CPU 和 IO 就不用分那么清楚了，合一起就行，怎么合呢，直接相加。

是不是比 Calcite 的默认处理要好，是。够不够好，不够。

前面那个小节都给了改进版的代价模型了，为什么还非得带上行数呢？CPU 和 IO 为什么不需要区分优先级呢？就算不用，合起来算也行，为什么要相加呢，为什么不是相乘，就算相加，为什么不带权重呢？

Spark 都知道搞个权重呢：

cost = weight * cardinality + (1.0 - weight) * size

这些问题，很难找到准确的解释。我们姑且可以这样理解：

代价模型是不完美的，总是在持续演进
行数被保留，一方面是历史原因，一方面正是因为代价模型的不完美，所以需要行数这个虽嫌不够直接，但好歹表达力丰富的变量来作补充
至于权重，是加还是乘，这些具体的算法固然会影响结果，但直接相加的方案可能实际效果也并不差了，胜在简单

除了 HiveVolcanoPlanner，CalcitePlanner 里还需要关注的就是 HiveDefaultRelMetadataProvider 这个类。

可以看到，除了默认的 DefaultRelMetadataProvider，还注册了一串在 hive.ql.optimizer.calcite.stats 中的类作为 MetadataProvider。回顾上面 Calcite 的架构图，也正是通过允许自定义 MetadataProvider，使得 Hive 能很方便的把 CBO 集成进去。