Spark SQL CBO 基于代价的优化

Posted 2023-04-15

参考技术A Spark CBO 背景

本文将介绍 CBO，它充分考虑了数据本身的特点（如大小、分布）以及操作算子的特点（中间结果集的分布及大小）及代价，从而更好的选择执行代价最小的物理执行计划，即 SparkPlan。

Spark CBO 原理

CBO 原理是计算所有可能的物理计划的代价，并挑选出代价最小的物理执行计划。其核心在于评估一个给定的物理执行计划的代价。

物理执行计划是一个树状结构，其代价等于每个执行节点的代价总合，如下图所示。

而每个执行节点的代价，分为两个部分

每个操作算子的代价相对固定，可用规则来描述。而执行节点输出数据集的大小与分布，分为两个部分：1) 初始数据集，也即原始表，其数据集的大小与分布可直接通过统计得到；2)中间节点输出数据集的大小与分布可由其输入数据集的信息与操作本身的特点推算。

所以，最终主要需要解决两个问题

Statistics 收集

通过如下 SQL 语句，可计算出整个表的记录总数以及总大小

从如下示例中，Statistics 一行可见， customer 表数据总大小为 37026233 字节，即 35.3MB，总记录数为 28万，与事实相符。

通过如下 SQL 语句，可计算出指定列的统计信息

从如下示例可见，customer 表的 c_customer_sk 列最小值为 1， 最大值为 280000，null 值个数为 0，不同值个数为 274368，平均列长度为 8，最大列长度为 8。

除上述示例中的统计信息外，Spark CBO 还直接等高直方图。在上例中，histogram 为 NULL。其原因是，spark.sql.statistics.histogram.enabled 默认值为 false，也即 ANALYZE 时默认不计算及存储 histogram。

下例中，通过 SET spark.sql.statistics.histogram.enabled=true; 启用 histogram 后，完整的统计信息如下。

从上图可见，生成的 histogram 为 equal-height histogram，且高度为 1102.36，bin 数为 254。其中 bin 个数可由 spark.sql.statistics.histogram.numBins 配置。对于每个 bin，匀记录其最小值，最大值，以及 distinct count。

值得注意的是，这里的 distinct count 并不是精确值，而是通过 HyperLogLog 计算出来的近似值。使用 HyperLogLog 的原因有二

算子对数据集影响估计

对于中间算子，可以根据输入数据集的统计信息以及算子的特性，可以估算出输出数据集的统计结果。

本节以 Filter 为例说明算子对数据集的影响。

对于常见的 Column A < value B Filter，可通过如下方式估算输出中间结果的统计信息

上述估算的前提是，字段 A 数据均匀分布。但很多时候，数据分布并不均匀，且当数据倾斜严重是，上述估算误差较大。此时，可充分利用 histogram 进行更精确的估算

启用 Historgram 后，Filter Column A < value B 的估算方法为

在上图中，B.value = 15，A.min = 0，A.max = 32，bin 个数为 10。Filter 后 A.ndv = ndv(<B.value) = ndv(<15)。该值可根据 A < 15 的 5 个 bin 的 ndv 通过 HyperLogLog 合并而得，无须重新计算所有 A < 15 的数据。

算子代价估计

SQL 中常见的操作有 Selection（由 select 语句表示），Filter（由 where 语句表示）以及笛卡尔乘积（由 join 语句表示）。其中代价最高的是 join。

Spark SQL 的 CBO 通过如下方法估算 join 的代价

其中 rows 即记录行数代表了 CPU 代价，size 代表了 IO 代价。weight 由 *spark.sql.cbo.joinReorder.card.weight *决定，其默认值为 0.7。

Build侧选择

对于两表Hash Join，一般选择小表作为build size，构建哈希表，另一边作为 probe side。未开启 CBO 时，根据表原始数据大小选择 t2 作为build side

而开启 CBO 后，基于估计的代价选择 t1 作为 build side。更适合本例

优化 Join 类型

Spark SQL 中，Join 可分为 Shuffle based Join 和 BroadcastJoin。Shuffle based Join 需要引入 Shuffle，代价相对较高。BroadcastJoin 无须 Join，但要求至少有一张表足够小，能通过 Spark 的 Broadcast 机制广播到每个 Executor 中。

在不开启 CBO 中，Spark SQL 通过 spark.sql.autoBroadcastJoinThreshold 判断是否启用 BroadcastJoin。其默认值为 10485760 即 10 MB。

并且该判断基于参与 Join 的表的原始大小。

在下图示例中，Table 1 大小为 1 TB，Table 2 大小为 20 GB，因此在对二者进行 join 时，由于二者都远大于自动 BroatcastJoin 的阈值，因此 Spark SQL 在未开启 CBO 时选用 SortMergeJoin 对二者进行 Join。

而开启 CBO 后，由于 Table 1 经过 Filter 1 后结果集大小为 500 GB，Table 2 经过 Filter 2 后结果集大小为 10 MB 低于自动 BroatcastJoin 阈值，因此 Spark SQL 选用 BroadcastJoin。

优化多表 Join 顺序

未开启 CBO 时，Spark SQL 按 SQL 中 join 顺序进行 Join。极端情况下，整个 Join 可能是 left-deep tree。在下图所示 TPC-DS Q25 中，多路 Join 存在如下问题，因此耗时 241 秒。

开启 CBO 后， Spark SQL 将执行计划优化如下

优化后的 Join 有如下优势，因此执行时间降至 71 秒

总结

5万人关注的大数据成神之路，不来了解一下吗？

5万人关注的大数据成神之路，真的不来了解一下吗？

5万人关注的大数据成神之路，确定真的不来了解一下吗？

分布式技术追踪 2017年第十九期

分布式系统实践

1. BigData－‘基于代价优化’究竟是怎么一回事？

http://dwz.cn/5Vq8T8

摘要: 很多同学应该都有认识, SQL的join操作很可能是性能杀手, 如果写的不好, 性能可能会大幅度下降. 很多SQL引擎都尝试进行join的优化, 其中基于代价的优化(CBO)是一种非常有效的手段, 这篇文章深入浅出的介绍了CBO的原理.

2. 从分布式到云端服务：Google Spanner 成长之路

http://dwz.cn/5VjE0v

摘要: 距离 Google 开始开发 Spanner 已经 10 年，5 年前 Google 发表了论文，在 Google 云平台上增加开放 Spanner 服务，意义不仅仅是服务于 AdWords 和 Google Play，而是希望在云端更有所为。在这 5 年时间里，正是由于其他厂商无法复制 Google 的想法，即不能解决大规模集群下高可用性、水平扩展能力、数据强一致性等。本文从 Spanner 技术着手，逐渐引出最近的特大消息：Google 开放 Spanner 云端能力（测试版）

微服务技术

1. 58到家立体监控平台：三大方面九个维度，架构流程及细节解析

http://dwz.cn/5VjGyJ

摘要: 这篇文章详细介绍了58到家的立体化监控平台, 实施一个全面的监控系统还是比较复杂的, 学学58到家的经验吧.

2. 详解如何设计私有云基础架构

http://dwz.cn/5Wc6UR

摘要: 本文非常全面的介绍了私有云在存储, 计算, 网络, 安全这四个方面的基础架构, 有助于大家全面理解私有云.

高可用技术

1. TCP接入层的负载均衡、高可用、扩展性架构

http://dwz.cn/5ULYCS

摘要: 这篇文章介绍了实施TCP层负载均衡实现高可用和可扩展的架构演变

2. 防雪崩利器：熔断器 Hystrix 的原理与使用

http://dwz.cn/5VNnYz

摘要: 本文介绍了雪崩的原理, 并且介绍了著名的开源熔断器Hystrix的原理和使用方法.

丰富多彩的计算机世界

1. Log:被BigData遗忘的奠基者

http://dwz.cn/5UaKJz

摘要: Log非常重要, 尤其是在大数据时代, 可以说是奠基性的技术之一. 本文对我们司空见惯的Log在大数据系统里面的巨大作用做一个推广和普及.

2. 伪共享，并发编程无声的性能杀手

http://dwz.cn/5UMcDV

摘要: 什么是伪共享? 伪共享是如何影响多核程序性能的? 这篇文章非常详细的介绍了伪共享的原理以及解决方法.