分布式环境中的连接计算和聚合计算（主讲人：王鹏飞）在线视频

所谓大数据上的关系型计算

最讲究策略的是连接和聚合

在不同的条件和要求下

会有不同的执行路径

尽管最后的结果都相同

好消息是这些策略

大多都有章可寻

有时候和同事们聊起来

每个人都在报怨

系统实现的太丑陋

因为我们永远都在

做各种各样的平衡

方案A顾得了内存

顾不了IO

方案B顾得了执行时间

顾不了

集吞吐量等等

但是我们后面的课程

会简化这些场景

只以IO优化为主

1TB的“产品表”

和3TB的“订单详情表”

在产品ID上的连接

这是一个真实的场景

并不是简单的YY出来的

实际上淘宝的产品表的大小

要比这个要大

我们假设“产品表”分布在

17台计算机上

而“订单详情表”大概分布在

60台计算机上

然后它们的这个数据分布

是没有特点的

数据分布没有特点是说

我们不能根据数据本身

去计算出来这条数据

在哪台机器上

然后在做连接的过程当中呢

它实际上是有一些悖论

比如说我们拆分的

计算节点越多

理论上作业完成的速度

就会越快

因为我们拆的越多

那么每一个计算节点

它要处理的数据就会越少

failover的成本也会越低

因为它计算的数据少

那么我们每次重新计算的

这个代价就会很低

但是计算节点越多

调度成本就越高

因为我们的计算节点有限

假如我们只有100个节点

但是我们预计需要1000个节点

才能完成所有的计算的话

那么实际上我们要调度10轮

所以为了在计算时间

和调度开销之间去达到平衡

我们设置每个计算节点

一次最多处理256M数据

而这是一个经验数据

我们实际上在控制

每个计算节点的计算时长

一般情况下

我们会把它控制在15分钟之内

对于这种数据场景

一般我们会选择Mergejoin

我们通过Shuffle

把来自“产品表”

“订单详情表”相同的ID

分到一台机器上去计算

然后通过sort

使进入Mergejoin算子的数据

满足Mergejoin需要的排序要求

按照“订单表”

“订单详情表”的产品ID有序

这样的话

它实际上会生成

三个不同的Task

Task和Task之间的边界

大家可以理解成进程边界

进程可能在同一台服务器上

也可能在不同的服务器上

一般情况下

我们划分进程边界

就是只看要不要

在集群上重新分布数据

大家请注意Task1

Task1它是去扫描“产品表”

然后呢

它在ShuffleWrite

operate（音02：32）里面

做的事情是说

Shuffle By ID

Sort By ID

这只是它逻辑上完成的工作

然后真正在物理上

它的这个执行方式可能是说

首先我按照ID分片

然后对每一片去做外部排序

然后排序完成之后

再把每一片的结果

写到DFS上

DFS是分布式文件系统的简称

它这样做是一种方式

另外一种方式

是首先做完全部的排序工作

然后再来分片

大家可以想一想

这两种方式

在执行上的这个不同

一部分是全局有序

一部分是片内有序

实际上它带来的算法复杂度

是完全不同的

Task2和Task1类似

只是说Task2扫描的是

“订单详情表”

它同样是按照products ID

去做Shuffle和Sort

Shuffle过后

相同的ID会在同样的

执行Task3的服务器上

并且因为我们前面做了

这个Sort

那么输入到Mergejoin

当中的数据

实际上已经是按产品ID

排过序的

后面就是标准的

Mergejoin的这个算法

在Task3中ShuffleRead

它做的是归并排序的操作

因为Task1和Task2

很可能会有很多的instance

对于products表Shuffle

出来的数据

如果Task1有10个instance

Task2有11个instance的话

那么Task3上面的ShuffleRead

它实际上要做10路的归并排序

下面的ShuffleRead它要做

11路的归并排序

这是一个很简单的操作

后面我们来看一看它的IO开销

首先我们看一下跨网络读

跨网络读是Task1

读取的数据量

加上Task2读取的数据量

再加上Task3读取的数据量

因为对于Task1来说

products表本身也是分布在

很多的机器上的

我们用最坏的方式估计

我们假设所有的读

全都是远程读

同理 Task2

然后Task3它的读取的数据量

是Task1和Task2写给Task3的

数据量的总和

所以最后我们计算出来是8个T

然后我们再来看

跨网络写

只有两个

分别是Task1和Task2

他们写出去的数据量

就是他们Shuffle的数据量

大概是4个T

然后本地的外部排序的开销

对于Task1来说最坏的情况下

它是写了1T

然后读了1T

Task2是写了3T

读了3T

我们把它们加到一起

本地外排的开销是8个T

大家可以注意到

本地外排的开销它已经

很接近于跨网络的IO了

所以说外排在分布式系统当中

也是一个巨大的开销

下面我们再来看另外一种

不同的数据场景

我们有一张2TB的“订单表”

它可能分布在几十台机器上

然后我们有一张100K的

“省份表”

然后我需要把这两张表

连接起来

按照省份ID

大家可以注意到

现在的数据的大小

和前面的数据场景

是完全不同的

前面是1TB和3TB去做连接

然后现在是2TB和100K

去做连接

然后它所使用的计算方式

也会不同

如图所示

我们会读取“省份表”

并且把“省份表”向每一份

读取“订单表”的worker

去广播

然后用“省份表”的数据

去建立一张Hash（06：41）表

然后用Hash（音）算法

去处理“订单表”的

每一条数据

这样的话

Task1实际上只有一个

instances

因为“省份表”这么小

我们只需要一份

instances去读就OK了

Task2呢

我们的instances数量

是2TB除以256M

然后我们再来计算

它的开销的话

是这样

我们首先来看跨网络读

跨网络读有两部分

一部分是Task1

去读的“省份表”

另外一部分是Task2

读取的广播过来的“省份表”

和“订单表”

大概计算出来是约等于2个TB

跨网络写只有Task1

在跨网络写

写的数据量是100KB乘以

Task2的instances的数量

大概只有400M

然后它实际上没有本地的开销

大家可以想一想

如果按照Mergejoin的算法

去做的话

IO量一共会有多少

前面提到的HashJoin的算法

它很特别

它特别之处在于说

我们使用province表

去建立那张Hash表

建立起来非常的容易

因为100KB我们非常轻松地

就放在内存里面

然后大家现在看到的

这个使用场景

2TB的“订单表”

和100GB的“用户表”

在用户ID上去做连接

想让100GB的表

就没办法轻松地放到

内存里去了

但是如果我们用最开始的

Mergejoin的算法的话

我们需要把2TB的“订单表”

去Shuffle Sort一次

把100GB的表

也Shuffle Sort一次

所以给人感觉

用Mergejoin算法

那么我们的本地IO的开销

会非常的吃亏

那么我们怎么做呢

我们像Mergejoin那样

我们按照用户ID去

Shuffle一次

但是我们不去做外排

也就是说

我们后面使用HashJoin的算法

然后我们给这种

只做Shuffle的HashJoin方式

取了一个名字

叫做Shuffle的HashJoin

这是我们生成的物理查询计划

Task1来说呢

我们把Users表

按照ID去Shuffle一次

Task2呢

我们把Orders表

按照Users ID也去Shuffle一次

Shuffle完的结果

因为我们把100GB

我们分成足够小的份数的话

那么这些我们是可以

放到内存当中的

尽管Orders表被Shuffle完

每一块还是会非常大

但是因为我们处理Shuffle表

是逐条处理

所以它不会有内存

被撑爆的状况

同时我们又避免了

在ShuffleWrite的过程当中

本地外排的开销

所以它比前面的两种方式

都要优一些

这是它的IO开销的估计

我们可以看到

Instance的数量的话

对于Task1来说

大概的是100G除以256M

对于Task2来说

是2TB除以256M

Task3我们把它设置成

和Task2的Instance数量一致

然后我们再来看

跨网络读有多少

大家可以简单计算一下

大概是约等于4TB

然后跨网络写是2TB

当然它没有本地的这个开销

大家可以想一想

如果用Mergejoin的算法的话

它的开销会有多大

这是另外一个非常典型的例子

我有2TB的“订单表”

然后我需要和100GB的

“用户表”在用户ID上

去做连接

我要需要再和40GB的

“卖家表”

在卖家ID上去做连接

然后我们可以比较一下

用前面的算法

我们会遇到什么困难

用Mergejoin呢

因为它存在排序

从本地IO上讲我们很亏

然后用HashJoin的算法呢

100GB老规矩

还是放不下了

然后如果用Shuffle的

HashJoin

那么实际上呢

我们在计算第一次连接

到第二次连接过程当中

我们需要把计算结果

再Shuffle一次

那么网络上Shuffle的数据量

又过于的大

那么我们有没有什么办法说

我们不Shuffle

那么大的数据量呢

然后解法很类似

我们的想法是说

我们保证最大的那张表

只Shuffle一次

然后那两张小表多拷贝几次

只要IO上

我们认为成本是合算的

那么我们就可以做

这是具体的算法

我们的做法是这样

首先把Orders表

分成Sellers表除以256M

那么多份

然后再乘上Users表

除以256M那么多份

然后它会形成一个

二乘二的矩阵

我们这一次的Shuffle

我们要把Orders表

分成一个矩阵

然后矩阵的行是Users ID

矩阵的列是Sellers ID

然后行的行数是

Users除以256M份

然后列是Sellers除以256M份

然后我们再来把Sellers表

分成Users除以256M份

并且把其中的每一份

广播到Orders所对应的

每一行上

然后再把Users表分成

Sellers除以256M份

然后再广播到Orders表的

每一列上

如图所示

然后在每一个矩阵上

我们就可以做HashJoin

它的做法的出发点

是通过聚融于Sellers表

和Users表

来避免这张大的Orders表

被多次Shuffle

它大概对应的物理查询计划

如图所示

实际上这张图画的相对来说

简单了一些

大家可以想一想

每一个BCRead

BC是broadcast的简称

BCRead它实际上

真正做的事情是什么

现在我们来看一下

这种连接方式

它的instances的数量

Task1是100G除以256M

Task2是40G除以256M

Task3是2TB除以256M

Task4的数量是Task1乘以

Task2的instances的数量

大家可以想一想

它的IO开销会有多大

然后可以据此来分析一下

这种连接方式在什么情况下

是合适的

在什么情况下是不合适的

大家可以看到

我们在前面的计算过程当中

严重地依赖Shuffle

但是Shuffle有可能造成

数据的长尾

比较典型的例子是

比如说少数大店的订单

它的这个数量一般会远远高于

比如说我个人的店铺

这个时候计算Orders

和Shops在ShopID上的连接

会出现绝大多数

instance已经计算完成

但是剩下几个instance的数量

由于要处理的数据量太大

执行时间非常长的这种状况

解法相对来说很简单

只需要把这些大店的数据

单独拿出来

然后通过HashJoin

去计算完之后

然后再合并到其他的

计算结果当中去

在这里顺带地一提

和JOIN有关的其他的优化

在这里就不详细地展开

给大家讨论了

大家可以用我提示的这些信息

去网上找一找

一个是和JOIN相关的

Predicate Pushdown

Join Reorder

这是在查询计划领域

相当困难的部分

另外就是在Shuffle数据的时候

NULL值的处理

前面我们在讲解连接的时候

我们发现就不同的数据场景

会需要不同的算法

实际上场景是说数据分布

连接 聚合也是一样

数据分布会决定不同的算法

比如说同样是对Orders表

进行聚合

我可能会计算每一个买家的

订单数量

我可能计算每一个类目的

订单数量

然后呢

我也可能计算每一个类目的

买家数量

最后呢

我们会说一下

对于计算每一个类目的

买家数量时

产生的数据倾斜

我们要怎么处理

计算每一个买家的订单数量

SELECT Buyer

COUNT（*）FROM

Orders GROUP BY Buyer

这是一条THIC（音00：48）语句

来表示它的计算逻辑

然后对于Orders表在

跟这条计算相关的

数据分布的特点

有这么两条

一个是买家是非常多的

内存里面我们肯定装载不下

另外呢

每个买家的订单数量

实际上都不多

换句话说

Buyer字段它在Orders表

当中的选择度是非常高的

一个可能的解法是这样

我们先扫描Orders表

然后我们按照Buyer这个字段

去做Shuffle跟Sort

达到的效果是

相同买家的数据

全部都到了一台机器上

并且它是按照买家有序的

这样我们在Task2当中

我们按照Buyer去排序

然后后面我们基于流

去做一个聚合

streamed的Aggregate呢

它的输入是按照Buyer有序的

这样的话

我们在真正的计算的时候

我们拿到买家A的第一条数据

然后依次地计算

一直碰到第一个

不是买家A的数据为止

这样的话我们就会产生一条

关于买家A的订单数量的记录

然后再去计算买家B

计算买家C等等

大家可以看到

Task1跟Task2之间

Shuffle的数据量的IO

是整个Orders表的IO

然后这件事实际上

是相当浪费的

我们来看一看

有没有什么优化的办法

优化的方式很简单

我们先做一次本地的聚合

它尽管Orders表

被分成了很多份

但是我在每一份上

先去看这一份内部

每一个买家他有多少个订单

这样话我们在往Task2

Shuffle数据的时候

我只要Shuffle这一份上买家

和这个买家的订单数量即可

这样的话我们会少Shuffle

非常多的数据

然后具体的做法是这样

我们在Orders表的

TableScan后面

补一个Hash-semi

aggregate

然后再去做Shuffle

ShuffleWrite做的事情

和前面一样

Shuffle/Sort By Buyer

然后后面的执行流程

和前面也相同

Hash-semi aggregate

它做的工作大概的流程是这样

首先它在它内部维护一张

哈希表

哈希表的key是Buyer

value是COUNT（*）的

Partial Result

之所以成为Partial Result

是因为我们在没有重新按照

Buyer去分布数据之前

我们是不可能拿到

最终的结果的

也就是说

它是一个最终结果的一部分

所以我们称为Partial Result

然后一条记录

我们从Orders表中

拿到的一条记录

我们首先去查这张哈希表

如果找到了相同Buyer的

记录的话

我们就在Partial Result上聚合

如果没有找到我们就插入

然后并且使用1来初始化

Partial Result

如果哈希表达到这个

它的容量上限

我们输出哈希表的10%

然后写到ShuffleRead当中

之所以是Hash-semi

aggregate semi的

意思就是说

并不是把每一个分片

做一个完整的聚合

然后再输出出去

也就是说

这里的Task1输出的记录

对于相同的Buyer来说

仍有可能会有多条记录

streamed的Aggregate

处理的是Partial Result

到Final Result

因为Task1到Task2之间

我们Shuffle的数据

不再是Buyer了

我们Shuffle的数据是

Buyer和它对应的

Partial Result

所以sreamed的Aggregate

它的输入大家可以看到

是ShuffleRead读过来的Buyer

和Partial Result（04：46）

然后streamed Aggregate

它在Partial Result的基础上

去计算出来Final Result

我们可以简单地去看一下

我们省了多少IO

比如说我们每一个买家

有三条订单

最好的情况下

所有的用户都可以放到

哈希表当中去

那么我们可以计算一下

Task1和Task2之间

Shuffle Sort的数据量

占原来的数据量的多少

实际上是三分之一了

计算每个类目的订单数据

类目字段是Category

如果用一条THIC（音）

来表达计算逻辑

是SELECT Category

COUNT（*）FROM Orders

GROUP BY Category

它的数据特点

是Category的个数非常有限

比如说衣服 鞋帽

电子产品等等

内存当中如果我们去放

Category数据的话

实际上是完全可以放得下的

但是每个Category订单数量

一般情况下都很庞大

比如说衣服

换句话说Category字段

它在Orders表中的选择度

是非常非常低的

解法和上例当中唯一的不同

在于我们用Hash aggregate

来代替掉了Hash-semi

aggregate

原因是Category个数有限

我们内存当中有一张哈希表

完全放得下

但是Buyer数据太大了

我们在内存当中放不下

另外的不同是

我们在Task1到Task2中间

我们在做Shuffle数据的时候

因为我们采取了哈希的算法

所以我们之所以Shuffle

不需要Sort

我们只需要按照Category

去做Shuffle好了

然后在Task2当中

我们使用Hash Aggregate

代替了上例当中的streamed的

Aggregate去做计算

我们再看下一个例子

计算每一个类目的买家

和卖家数量

用一条THIC（音）表达

SELECT Category

COUNT（DISTINCT Buyer）

COUNT（DISTINCT Seller）

FROM Orders

GROUP BY Category

我们再来看针对这条语句来说

它有什么样的数据分布的特点

首先Category的个数有限

这是一个好消息

但是买家和卖家的数量很多

内存还是放不下

甚至在有些情况下

一个Category下的买家和

卖家内存当中也放不下

比如说像衣服 鞋帽

这种非常巨大的Category

买家他会在多个Category下

买入商品

也就是说我去淘宝买东西

很可能会买电子产品

也会买衣服 鞋袜

然后每个Category的

买家和卖家数量差距会非常大

举个例子

比如说汽车和服装

买车的人应该很少了

针对这样的数据特点

我们影响去怎么处理呢

第一种解法

很自然

就是我们先求买家数量求出来

再把卖家数量求出来

然后我们把结果连接起来

形成最终的结果

比如我们计算买家数量

我们会用这么一个查询计划

同样是扫描Orders表

然后我们维护一个

Hash-semi aggregate

这个key是Category和Buyer

value我们什么都没有

换句话说

这个Hash-semi aggregate

它做的事情只是在本地去重

然后我们按照Category

去做Shuffle

然后我们按照Category

和Buyer去做Sort

然后在Task2当中

我们按照Category和Buyer

去做归并排序

然后这样话

我们能拿到按照Category

和Buyer已经排序过的记录集

然后我们按照streamed的

Aggregate去做这个聚合

我遇到相同Category下的

Buyer的话

那么我只计算一次

直到遇到不同的

我再相加

然后计算卖家的数量类似

然后在我拿到最终的结果之后

我们可以以Category为key

然后去做连接

然后整个的计算过程

用一条THIC（音）来表达的话

基本上是这个样子

大家可以看里面的两个子查询

我们先计算出Orders表当中的

每个Category下面的

卖家有多少

再计算出来买家有多少

然后再去做一个连接

现在我们来分析一下

这种解法的开销

首先Orders表被读了2次

一次是用来计算买家

一次是用来计算卖家

然后Category在Shuffle

Sort的时候

实际上它在Streamline上

被传输了两次

然后连接是非常重的算法

它的计算开销和其他算法来比

相对来说会大一些

如果Category它自己的

选择度而比较高的话

换句话说

如果Category个数非常多的话

那么连接的开销很可能

是不能忽略的

另外呢

这个解法它没有去解汽车的

买家很少

但是服装的卖家很多的问题

换句话说

我们按照Category去做Shuffle

它会有非常严重的长尾问题

那么我们怎么改进呢

在这里我们要利用

聚合函数的一个性质

聚合函数它的输入参数

为NULL的时候

它会忽略对当前行的计算

所以我们可以先膨胀数据

然后再去做聚合

具体怎么做呢

就是这样

我们老规矩

先去扫描Orders表

然后在Hash aggregate的时候

我们分成两步做

第一步先去膨胀

后面我会讲膨胀是如何做的

然后以Category Buyer

和Seller为key去去重

然后后面再去Shuffle

Shuffle的顺序是按照Category

去做Shuffle

然后Sort是按照Category

Buyer和Seller去做Sort

后面的处理流程跟前面很像

我们先来看

我们是怎么做膨胀的

这是一个膨胀的例子

对于左边表的每一行

我们会把它扩展成

右边表的两行

然后扩展的时候

Seller和Buyer对应的这个字段

我们分别替换成NULL

比如对这一行来说

第一行的数据是服装

阿飞和黄蓉

它扩展成两行

会变成服装 阿飞 NULL

服装 NULL 黄蓉

其他行类似

然后我们在Sort By Category

Buyer和Seller的时候

我们把NULL值作为最小值

去处理

那么按照Category Buyer

Seller排序完

排序的结果如右图所示

大家可以看到

在右边这张表

它Buyer会是连续的NULL

Seller也会是连续的NULL

前面我们提过说

聚合函数它在遇到参数值为

NULL的函数的时候

它会忽略这一行的计算

这就意味着说

COUNT DISTINCT Seller

COUNT DISTINCT Buyer

这两个聚合函数

分别只在右面数据集的

部分行上生效

而这样的话实际上

我们可以在一次计算过程当中

把两个值全部都计算出来

这样就躲过了Join

然后再来看我们后面

在Sort完之后

我们是怎么去重的

去重很简单

我们把左边的记录变成了

右边的记录

大家可以看到

服装 黄蓉这两条记录

变成了一条

然后我们再来看一下

对解法2的分析

首先Orders只被读了1次

Category在Streamline上

只被Shuffle了1次

这都是我们得到的好处

然后没有连接了

但是坏消息是

我们依然没有去解决

Category数据量太突出的问题

长尾的问题

然后我们

我们怎么去解决长尾问题呢

我们这样

中间多加一次Shuffle和Sort

我们先来看Task1

TableScan和Hash-semi

aggregate

这两个operate（音13：18）

做的事情和解法2是完全相同的

但是我们在Shuffle的时候

不是按照Category去做

Shuffle的

而是按照Category Buyer

和Seller去做Shuffle

这样的话我们能保证

相同的Category Buyer

和Seller的数据

会到Task2的一个instance上去

然后我们再按用streamed的

Aggregate

去在Task2当中去重

实际上在Task2当中去重

除了去重之外

它还要去做一些其他的事情

然后我们还需要另外一个阶段

Task3

大家可以根据我们前面的讲解

来想一想

Task2和Task3

我们要完成什么样的工作

解法3和前面的解法不同之处

在于它多了一个去重的阶段

专门去重的阶段

Task2它实际上出的是

Partial Result

在Task3当中

我们继续要计算

Final Result

然后这个时候

大家可以去想一想

Task2当中的Partial Result

应该是什么样的partial

这个问题留给大家

现在我们来分析一下解法3

首先Orders表只被读取了1次

但是我们多了一次Shuffle

Category因为被Shuffle了2次

它实际上占用了集群的

额外资源

然后没有连接

之所以说没有长尾

有两个原因

一个是Task1到Task2之间

我们是按照Category

Buyer和Seller去做的Shuffle

然后他们Shuffle出来的

数据结果应该是相当均匀的

另外Task2到Task3之间

尽管我们会按照Category

去做Shuffle

但是Task3的输入

尽管还会有长尾

但是那个尾巴的长度

会和Task2的instance的

数量有关

也就是说

它哪怕长也不会长到哪儿去

是我们可以接受的

对于解法3来说

在大多数情况下

我们会加速单条查询的

执行时间

但是我们实际上损害了

集群的吞吐量

在集群负载很高的时候

多这一次Shuffle

实际上对于集群的运算能力

是一种损害

这也是我们前面提过的

我们需要做的各种平衡之一

我们前面介绍了聚合和连接

但是实际上用户总会把

所有的操作用到一起来用

比如说大多数情况下

都会有连接

都会有聚合

可能会有UNION ALL

可能会有窗口

但这些计算混在一起的时候

我们如何计算得更快呢