Apache Hive 如何进行基于成本的优化?
上一篇文章 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 集成进去。
如上图的 HiveRelMdSelectivity,就通过 getSelectivity() 这个方法定义怎样去计算一个 Filter 的 Selectivity,而这正是是计算 Filter 操作代价的核心。
限于篇幅,就不再跟踪更多的代码细节了。恐怕大部分人也没兴趣深入。到这里,这篇文章就结束了,来总结下。
- CBO 相较于 RBO,是一种更加准确和高效的优化方法
- Hive 通过 Calcite 灵活的架构,很方便的实现了 CBO
- 需要明智的收集足够的数据分析结果来帮助 CBO
- Hive 的代价模型还不够完美,至少需要更好的 cost function 和 准确的 histogram
推荐阅读
-
Apache Hive 如何进行基于成本的优化?
-
小红书大产品部架构 小红书产品概览--经过性能、稳定性、成本等多个维度的详细评估,小红书最终决定选择基于腾讯云星海自研硬件的SA2云服务器作为主力机型使用。结合其秒级的快速扩缩、超强兼容和平滑迁移能力,小红书在抵御上亿次用户访问、保证系统稳定运行的同时,也实现了成本的大幅降低。 星海SA2云服务器是基于腾讯云星海的首款自研服务器。腾讯云星海作为自研硬件品牌,通过创新的高兼容性架构、简洁可靠的自主设计,结合腾讯自身业务以及百万客户上云需求的特点,致力于为云计算时代提供安全、稳定、性能领先的基础架构产品和服务。如今,星海SA2云服务器也正在为越来越多的企业提供低成本、高效率、更安全的弹性计算服务。 以下是与小红书SRE总监陈敖翔的对话实录。 问:请您介绍一下小红书及其主要商业模式? 小红书是一个面向年轻人的生活方式平台,在这里,他们发现了向上、多元的真实世界。小红书日活超过 3500 万,月活跃用户超过 1 亿,日均笔记曝光量达 80 亿。小红书由社交平台和在线购物两大部分组成。与其他线上平台相比,小红书的内容基于真实的口碑分享,播种不止于线上,还为线下实体店赋能。 问:围绕业务发展,小红书的系统架构经历了怎样的变革和演进? 系统架构变化不大,影响最深的是资源开销。过去三年,资源开销大幅增加,同比增长约 10 倍。在此背景下,我们努力进行优化,包括很早就开始使用 K8S 进行资源调度。到 18 年年中,绝大多数服务已经完全实现了容器化。 问:目前小红书系统架构中的计算基础设施建设和布局是怎样的? 我们目前的建设方式可以简单描述为星型结构。腾讯云在上海的一个区是我们的计算中心,承载着我们的核心数据和在线业务。在外围,我们还有两个数据中心进行计算分流,同时承担灾备和线上业务双活的角色。 与其他新兴电子商务互联网公司类似,小红书的大部分计算能力主要用于线下数据分析、模型训练和在线推荐等平台。随着业务的发展,对算力的需求也在加速增长。
-
金融科技的高效省力秘籍:打造全面连接、全景覆盖、智能化的数字化运营体系" - 当下金融科技运营:挑战与机遇共存的时代解读 在快速发展的数字技术和企业数字化转型的大背景下,中国金融科技产业步入了提质增效的新阶段。面对市场的起伏变革与不确定性,金融机构需积极拥抱创新,灵活运用新技术,确保在竞争激烈的市场环境中稳固立足。 - 面临的双重考验: 1. 技术迭代压力:持续跟进行业内的科技革新,掌握新兴工具和平台,时刻应对瞬息万变的市场需求是金融科技运营的一大挑战。 2. 安全与隐私挑战:伴随着网络安全风险加剧和数据泄漏频发,如何强化信息安全体系、防范攻击、维护客户资金及隐私安全显得尤为重要。同时,伴随金融科技公司崛起,个人隐私权保障愈发关键。 - 喜人的发展空间: 1. 提升运营效益与降低成本:借助数字化技术,实现流程自动化、信息整合以及数据分析等,有效提升工作效能并缩减运营成本。 2. 扩大市场份额与增收途径:利用数字化手段拓宽销售渠道,优化用户体验,吸引更多用户并带动收入增长。 3. 加强客户联系与提升满意度:通过数字化科技运营,企业能更好地与客户互动沟通,增强客户信任感与忠诚度。 - 构建金融科技降本增效的核心驱动力:实施“全感知、全链接、全场景、智能”的科技运营体系升级路径