统一的计算框架（主讲人：强琦）在线视频

从上面的介绍可以看到

历史上的不同计算系统

目前大有融合的趋势

我们在这方面也进行了

非常多的尝试

下面 我会跟大家介绍一下

统一计算框架方面的心得

我们从统一计算框架的

视角来看

我们换一个纬度看问题

分成数据是不是分批执行

数据的Shuffle方式

是push式还是pull

数据进程或者说

用户task

是否需要预先拉起

分成这三个纬度

如图所视

我们可以看到不分批

以pull的方式

不预先拉起

那就是传统的离线

而不分区

采用push模型预先拉起

则是service mode形式

就是预先拉起

而分批 push模式

预先拉起则是传统的流计算

当然 我们支持这8种

任意组合

只不过有些组合下

效率比较低

引入一个状态计算的

计算框架 MRM

提供一个灵活高效的

Shuffle service

以及一个灵活的

APPMaster体系

可以看到 我们在进行

尝试的统一计算框架里

里面有很重要的角色

叫Runtime Controller

其中 有Local AM

它的意思是说在线的请求

或者对延时较高的请求

可以不唯一

通过AM来提交任务

每个Runtime Controller

里面

有一个本地的Local AM

保留着足够供它决策的信息

我们这里的Session概念

指的是在Session内的

所有类型的job

都可以复用数据 所以

Session是复用数据的边界

在Session之外的job

数据复用

只能使用 借助第三方存储

比如说DFS

或者类似与Tachyon

这样子的文件类型系统

或者其他的TV的存储

那么 在Local AM里头

有一个重要的管理组建

叫做DAGManager

里面一个重要的数据结

DAGSession

它是管理提前拉起的

物理执行dag

而与之对应的BlockManager

里面的重要数据结构

BlockSession

管理任务间

复用的数据的全局

运行时管理

也就是说 这个数据的

分区的哪个版本

和哪个replication

是在哪个进程 全局管理

而VertexManager

管理和开放

所有的控制逻辑运行时机

LocalAM保留着跟AM里头

数据结构一致的本地AM

包括DAGManager

BlockManager

和VertexManager

然后AM会通过心跳

将新增加的DAGManager

里头的信息

Block信息

同步给所有的

Runtime Controller的AM

localAM

所以 在线请求

提交任务的时候

如果发现localAM里头

已经有与这个请求相匹配的

拉起的物理的DAG

执行资源的话

直接选取这些资源

然后再查看

这个计算涉及到的数据

是不是已经被加载在

BlockSession里头去

如果已经有

进行合适的摆放

选择完所有所需要的资源后

直接下发Worker

当然 大家可以看到

不同的Runtime Controller

看到的资源是一样的

所以 与传统的离线不同的是

在Worker端会存在

不同任务的竞争

所以 在Worker端

会有本地的调度

我们算子层也提供了

五类基础算子

这五类基础算子正交

并且合计为全部的表示能力

那么 在这个基础算子之上

可以实现典型应用

向上派生出高级算子

可以复用

这是几个算子之间

数据结构的流转图

这里就不详细展开

可以看到这个Case

其中 用户的Case包含了

离线计算 实时计算

和adhoc的查询

这段代码是为

在线准备物理拓普

一旦拓普建立

在线请求将不会通过AM

而直接通过本地Local AM

找到合适的计算资源

进行下发执行

那么 这个是在线部分

也就是说 大家可以看到

在线的这部分query

直接可以复用离线拉起的

DAGSession

从而达到秒级以内的

在线请求

当然 这个也是

OLAP的请求

刚才我们也提到

统一计算框架引入

定义了Runtime Controller

其看到的资源基本一致

所以 需要Worker层的

本地调度

它具有灵活的表示层

但是 灵活的表示层

所有的数据类型都是范型

范型就会有较高的运行时

和内存的代价

所以 我们会引入对象池

以及内存池

来尽可能的缓解这个问题

Table是一个具备

schema的存储表示

所以 我们可以利用

用户schema

和系统schema

做大量的本地化的

物理计划执行优化

和算子的改写

比如说我们发现

两张Join的表

其实已经都加载在内存里头

而且支持组建的索引

而这两张表的partition

恰好与Join的key

是一致的

而这两个表又放在相同的

机器上 进程里

我们于是把这个Join的算子改写成LocalJoin

不需要任何的Shuffle

本地进行Join

即可得出Join的结果

我们也同时支持

LocalDataSet

即用户可以在自己的

控制代码里

new出任意的数据结构

将这个数据结构partition

传送远端 参与计算

同时 Dataset定义了版本

Partition Relication以及Tag

用户可以指定不同的Dataset

具备相同的Tag

系统在加载的时候

尽量使得相同Tag

摆放在相同的内存

我们定义了一些抽象行为

来指导系统的优化器

比如Match接口

系统将会问 这个Dataset

支不支持在B字段上的

大于操作

那么 如果Dataset的

Implementation

返回支持

实际上 也就是说

一个while语句A大于6

如果不支持这样的操作

物理执行计划

将会在scan上面插入一个filter操作

而如果Match通过

也就是说 该Dataset

支持在A字段上的

大于的 选择操作

那么 整个执行框架

将会被改写成

A大于6

将会下推到scan

读出来的结果就是

A大于6的结果

而无须scan所有的数据

过滤掉A大于6的结果

我们也引入了

CBO和RBO的框架

来进行Join

和大量的Pushdown

我们将不同的运行方式

与用户的处理落实剥离出来

使得用户的处理逻辑

只跟他的功能相关

而运行的Contex不同

离线的任务 Streaming

迭代 准实时 实时

在线 图分析

用户所表达的逻辑是一样的

只需参数的不同

用户的代码无需修改

即可运行在不同的Contex下

那么 对于在线请求

我们可以提前建立Dag

提前拉起

并且在线的SQL

或者在线的请求

Pattern是固定的

我们进行大量的优化

进行执行计划的下推

进行大量的Codegen

本地优化 内存优化

长尾控制 本地调度

以及截断测量

相比于跟Spark的区别

Dataset是可以进行修改

BATCH间进行串型DAG

BATCH间进行并行DAG

Shuffle service的方式

不对计算逻辑产生任何影响

任务跟踪方式进行统一

从Shuffle Framework

改进到Shuffle service

预先拉起进程

和预先task逻辑

变成一个运行方式

引入BATCH的逻辑

进行纯粹的内存计算

也就是说

不管数据量是不是超过

内存的边界

逻辑上都可进行

完全的内存计算

正如前面所介绍到的

这些主要是依靠

引入了增量的

有状态的计算模型

我们以上讨论了很多

可以看到数据库技术

是不是可以和BigData

分布式计算技术

进行有效的融合

和相互借鉴

数据达到不拖动

更好的Schema控制

更精细化的索引元数据

本地调度策略

CBO框架

以及引入更多的目标客户

这些目标客户已经很熟悉

SQL的语意和语法

那么 前面也提到了

我们在代码优化方面的

一些工作

从向量优化方面来说

我们可以

因为我们有了schema

我们就可以进行列式存储

列式存储很明显的好处

是可以做压缩

高效的压缩

当然 要做行的恢复

另外 我们可以利用CPU

单指定级多数据级的特点

如图所示

如果我们向左边的去运算

每次都会产生一条CPU指令

而优化成右边这样子

我们知道C0和C1是集中存储

在一次迭代中

实际上一次指令

可以把多条数据进行集中运算

相比于左边

这样子的

在密集数据计算情况下

可以把性能提升4到10倍以上

当然 只限于数据密集型

但是 指令类似

大家都知道在举证运算

和图形渲染方面

GPU主要就利用了这些特性

当在Codgen方面

大家都知道现代CPU的技术

分支预测和执行估计

本质上运行时

会出现比较大的问题

尤其是在switch/case

if/else

for/while这样的环节

使得整个分支预测实施外

那么 我们的核心要点

是将运行时的不确定性

变成运行时的确定性

所以 利用了Codgen技术

将序列化分序列化

因为大家很清楚

当我的格式确定的时候

我的反序列化

序列化的具体的字段

运行时是固定的

我们将在编译时

利用Codgen把这个代码

直接生成

减少CPU的分支预测的失败

包括虚函数

SQL的表达式

A大于5 大于6

因为我们在提交任务的时候

这个已经决定了

我们就不让它在运行时

用状态机的方法解决

而在编译时

就把它Codgen出来

包括DAG的执行

DAG的前区后区

如果在本地化

我们就直接已经确定了

就直接Codgen

另外 包括String优化

这个都可以做大量的

将运行时的不确定性

变成确定性 来进行

使得整个CPU更加友好

cache line也更加友好