其他计算和物理优化（主讲人：王鹏飞）在线视频

这部分会首先简介一下

相对简单的基于窗口的计算

然后聊一聊工程实践上

一些考虑的点和优化

关系型计算当中对窗口的定义

大多数是固定窗口

我们这次讲解

以固定窗口为主

窗口函数它有几个要素了

第一呢

如何将数据集分成窗口

换句话说

就是我们要在计算的时候

如何去shuffle数据

窗口内的数据

按照什么东西排序

这东西对应一个Sort

然后窗口上如何计算

这就是窗口函数

本身的计算逻辑

下面这条语句是一条典型的

窗口函数的计算

SELECT ROW-NUMBER（）
OVER（PARTITION BY
Cagegory ORDER BY Date）
rank FROM Orders

这句话的意思是说

首先这个窗口

是按照Cagegory去划分的

也就是说

每一个Cagegory划成一个窗口

然后窗口当中的数据呢

是按照Date去排序的

然后排序的结果呢

对每一个窗口当中的

每一行数据

依次的为它附一个值

从1开始

然后以1为步进去递归

以1为步进去自增

然后他生成的计划非常简单

一个TableScan

一个（00：50：59）

一个shuffleWrite
一个shuffleRead

和一个

Window（00：51：02）的计算

后面我们会介绍一下

在优化过程当中

我们需要考虑的点

首先呢

这个IO是主要目标

如果能减少shuffle的数据量

我们就减少shuffle

如果能避免不必要的

shuffle-Sort

我们就要避免不必要的

shuffle和Sort

实际上在我们生成

物理柴恩计划的过程当中

每一个Physical Operato

它自己都有shuffle-Sort属性

如果一个Opernto满足了

它对输入的要求

那么就可以不用去

shuffle或者Sort

比如说

shuffle By（a）

和shuffle By（a，b）

实际上前者兼容后者

Sort By（a，b）

和Sort By（a）

实际上前者也兼容后者

但是一般情况下

shuffle的K的数量减少了

那就意味着出现

长尾的可能性变大

因为shuffle By（a）

基本上不会比

Shuffle By（a，b）

能够分布的更均匀

这个时候

如果我们要去掉长尾的话

像前面我们在介绍

聚合时候提到的那样

我们需要在

单条查询的执行速度

和集群吞吐量上去

找到一个平衡

我们需要知道什么时候

我们应该为单条查询的

执行速度

去生成查询计划

什么时候我们要

以集群吞吐量为先

后面我们会简单介绍一下

一些典型的物理优化场景

比如说

A JOIN B ON A.AD=

B.ID GROUP BY A.ID

它大概意思是说

我有一条查询

我需要A跟B先在ID字段

上去做连接

然后呢

后面呢

我会在A的ID字段上

再去做一个聚合

这个时候实际上

因为连接它输出的数据

按照ID已经做过Shuffle了

那么我们在计算聚合的时候

我们不需要再去Shuffle一次

我们直接可以从原始结果

输出最终结果

这是一种场景

我们再来看另外一个例子

JOIN Reorder

A和B连接

在A的ID和B的ID列上

再和C连接

在A的CT和C的CT上

再和D连接

A的ID和B的ID上

如果我们认为

AB的结果和C连接之后的结果

会产生一个非常巨大的结果集

那么再连接D的话

实际上会有额外的开销

我们可能会调整

和 C D 的连接顺序

来避免产生非常大的结果集

在分布式场景当中

因为我们处理的数据量很大

我们需要尽可能的减少

我们扫描的数据量

列存储和列裁减

是另外一项非常重要的优化

在大数据处理领域

IO是最重要的一个优化目标

我们通过计算一条查询当中

用户所引用的必要的列的列表

来去从存储系统当中

只获取

必要的数据

来减少对存储系统的IO

列存储是说对一张表的数据

我们以列的方式存储

我们只获取以列A的数据

我们不需要把整行的数据

全部都拿出来

我们只需要列A的数据即可

Predicate Push-Down

在前面讲到JOIN的时候

我们提过一次

尽量可能早的去做
（00：54：35）

来减少在

（00：54：38）

后面的计算过程当中

计算的数据量

Partition Pruning

Partition

它本身是来自于

一个HAVE的概念

它从逻辑上讲

实际上是说

我们允许用户

对数据子集去打一个标记

查询的时候呢

它可以依据这个标记

来去选择它要查询哪些子集

以此来做到

只扫描部分数据

而不是扫描整张表

详情大家都可以去网上找一找

后面我会讲一讲

工程方案当中

一些常见的这个细节

首先呢是基于开销的优化

如果有数据分布的话

我们就会知道

Shuffle BY（a，b）

被降级成

Shuffle By（a）

会不会造成长尾

一个典型的例子是

Shuffle By（c t）

Users如果降级成

Shuffle By （c t）

这个非常有可能会变成长尾

因为青海的用户数量

一定会远远的小于广东

但是如果是

Shuffle By Users
（00：55：46）

如果降级成

Shuffle By Users

这个尽管有可能会有长尾

但是那个长尾的那个现象

会非常的轻微

比如说

最长的那个尾巴

和最短的那个尾巴

很可能它们只相差

百分之零点一

如果有数据分布的话

实际上我们还可以做

非常非常多的这个事情

数据分布在

传统关系型数据库当中

一般情况下

是用直方图来表示的

但是在海量数据的情况下

我们会遇到一些困难

比如说

在1T的数据上

如何生程直方图

但是不产生

过高的运行时开销

这是需要想一想的

另外呢

直方图本身

是不可能非常巨大

比如说1G大小的直方图

我们在生成查询计划

的过程当中

读直方图

计算直方图本身

它的开销很可能

一二十秒就已经过去了

这是不可接受的

换句话说我们需要知道

说如何让这个直方图

采样够小

但是呢

它足够有效

这都是我们面临的这个

工程方面的问题

另外呢

在工程实现方面

我们也会面临其他的状况

以我们看到的场景

我们现在是海量的作业

再加海量的数据

换句话说

作业数会非常非常的巨大

数据呢

也会非常非常的巨大

但是呢

只有极少数的大型作业

在一个作业当中

处理非常大量的数据

换句话说

这海量的作业当中

绝大部分作业

都是在三十秒以下的小作业

对于我们来说

一件不可思议的事情是

存储是瓶颈

我们的集群

经常会因为存储资源不够

而扩展

但CPU不是

然后有些用户的作业呢

时间比较敏感

比如说广告推荐

它一定要在

一个足够短的时间段之内

完成所有的计算

否认的话就会影响

推荐的质量

在这种时候

集群的吞吐量是退居其次的

我们所面临的工程现实

还会有非常多的这个复杂状况

都需要一个一个的

具体问题具体分析了

后面我们来看一看

我们的一些工程方案

首先呢

我们在执行引擎实现的细节上

在做不断的改进

然后

大家可以仔细去观察观察

SQL的计算逻辑

它大多是什么样子的

就是SQL这东西

它大多是按行计算

但是呢

它在固定的计算当中

它只引用固定的列

比如说

下面这条语句

SELECT col1+col2 FROM

Dual WHERE col3大于10

它计算的时候确确实实是以行

一行一行的去计算的

但是它只引用了

col1 col2 col3 这些列

并且呢

对于SELECT列表中的加法运算

它只引用了col1和col2

这两个列

在（00：58：54）当中

它只引用了col3 这一个列

所以一切优化

我们都围绕着SIMD走

SIMD它的全称是

单位指令多数据

这个大家可以简单的去

百度一下

然后为此呢

我们把内存中的数据

也摆成列式存储

来去充分使用SIMD

它会极大的增加

CPU Cacth的命中率

我们自己作过的内部测试

它比传统的行内存的解决方案

速度大概提升的百分之一百

我们还会有一些其他的优化

比如说

查询计划缓存

我们观察到了

用户每天提交的作业

除了常数参数不同之外

其他都相同

常数参数是什么呢

比如说

用户它总是在

查询前一天的数据

和当天的数据在做比较

今天呢

它查的是昨天和今天的数据

明天呢

它计算的一定是

今天和明天的数据

像这种数据

我们是可以把查询计划

缓存起来

来避免每次用户提交一条查询

我们都去生成查询计划

另外呢

我们还会提取公共的计算

公共计算是说

用户它会提交一系列的查询

来完成一连串的这个需求

我们可以

一次处理一整批的SQL

然后这样的话呢

一个是说

我们可以在非常广泛的

范围之内

去做这个源表合并

源表合并是说

如果一张表

它在查询的过程当中

被读取了多次的话

那么我们可以只从DFS当中

读取一次

来避免跨网络读

然后呢

有一些被反复使用的纬度表

我们可以让它常驻内存

来避免在加载的时候

需要读磁盘

我们通过合并一整批的SQL

我们有可能生成非常巨型的

单次作业

调度系统的开销

它和作业数量是成比例的

通过减少这个作业的数量

我们是可以减少

调度系统的开销的

当然这里也有问题

就是

我们在合并成一个

大的作业的时候

如果这个大的作业执行失败

那么我们需要重跑整个作业

重跑的这个时间开销

基本上和我们合并的作业

的规模是成正比的

所以我们需要在

减少调度系统开销

和作业失败

fanl over之间

去做平衡