273bb3f6726145fc17958da1c0d1957f
Apache Hive 是怎样做基于代价的优化的?

上一篇文章讲了 Apache Calcite 为什么能这么流行。今天我们看一个实际的应用,来聊下 Hive 是怎么利用 Calcite 做基于代价查询优化的。

基于代价的优化器

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

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

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

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

BO 固然是好的,能解决很多问题。

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


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

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 的数量呢?

top Created with Sketch.