有状态计算、并行DAG、抢占式调度和资源隔离、Failover机制（主讲人：强琦）在线视频

其实之前我们刚才一直在介绍

增量系统

它是一个有状态的计算

也介绍了我们这个

MapReduceMerge的模型

大家可以看到

Map和Reduce

我们约定在Batch内的一个

Map操作和arguate的一个

reduce操作

这个的语意和MapReduce

完全兼容

Merge我们可以看到

这个oldValue

就是我们刚才

一直保存的那个状态

这个是系统来帮助用户来保存的

用户只需要在Merge里面

去写用户逻辑

用刚才说的count去加就完了

oldValue的存放

全部是由增量计算框架来维持

输入这个Value

是Reduce

本次Reduce的结果

用户只用把oldValue

和这个value进行合并操作

并且返回新的值

作为下一次的oldValue传

当然大家可能会有疑问

countSum这个是可以增量的

可加的

但是有些操作

比如说distinct

它不可能有两个值

直接处理得出

所以整个计算框架

向用户提供了一个state

也就是说

用户可自定义的一个数据结构

那么既辅助这个数据结构

用户可以将这些问题

变成一个可增量

也就是说distinct

当你要求严格精确

你就可以去在这上面

定制一个bigset或者bigmap

当你允许

一定程度上数据的精度

节约系统开销 内存开销

那么你可以采用？

或者技术估计的方法

而把这个数据结构

绑定在这个State上面

系统来帮助你来做fillwork

也就是说

你不用去管整个状态的持久化

不用管它的容错等等

有的同学就有疑问了

那我这个state

通常地像Storm

以Kinesis和MillWheel

这个存储

它都是借助外部第三方的

可靠的持久化存储

比如说Hbase OTS

OTS阿里云的产品

类似Hbase这样的存储系统

它的读写

随机读写的性能

它就不能线性扩展

当我这个作业的并发数

增长的时候

到一定程度上

我的任务的并发程度的增长

已经受限于Hbase的随机

尤其是随机读的能力

所以对整个系统的scale ability

是有限制的

那么我们解

这个Scale ability的问题

引入了一个内存的

一个增量的snapshot机制

也就是说用户可以指定

固定批次的数据

在这个批次内的

对于刚才的OldValue

和state的修改

完全都在一个增量的一个snapshot

而这个snapshot在内存里头

当然内存不够用的时候

会持久会刷到SSD上

所以用户在这里

对这个值的修改

注意是update

Merge是update

它完全都是在

增量的snapshot内完成

而系统这时持续地引进

会产生大量的增量的snapshot

这时系统会在内部启动

一个checkpoint的线程

它会顺序地

将这些snapshot

选择性地进行Compact

Compact后

将这个内存的snapshot持久化

批量地刷入到盘古

刷入到一个快存储

全局的快存储

可以看到

这样子的机制

既保证用户在调用Merge

这个函数的时候

基本上都在操作内存

而整个的系统的scale ability

不依赖于任何的

也就是说这里头的读

随机读不依赖任何的

第三方的可靠存储

而系统将snapshot的checkpoint

是在异步的在后台进行

利用了第三方存储的可靠性

和吞吐

当然可能看完觉得

是不是很像LSM

事实上

它就是一个变种的LSM

而这个变种的LSM是存储

跟计算一体的

也就是说

系统在插入到snapshot

进行snapshot更新以后

snapshot里面内容的更新

它无需进行reload存储

而这个是在

无论是RocksDB LevelDB

都是做不到的

因为存储就是存储

它的语意insert进来

就应该是可靠的

所以它不可避免的

要去写穿readlog

并且要利用？等协议

进行repetition

一致性的repetition

而我们这里面的

一个弱的LSM

它是跟计算绑定在一起的

也就是说

我如果这个snapshot

还没有经过checkpoint

就宕机的话

这个incremental snapshot

会由

无论是源头

还是上游结点去重建

Rebuild

系统整个宕机后

会从最近的一个checkpoint恢复

重新加载

并且由最近的snapshot

checkpoint以后

会Replay数据

来重建内存里头的snapshot

这里需要再说明一下

就是当前只存在一个snapshot

是可写可读

当这个snapshot关闭以后

就固定Batch以后

这个snapshot就变成一个（）

所以整个机制是一个lockfree

可以做到非常的高效

为了克服datascale

以及增加系统的时效性

所以我们整个DAG

完全是一个并行的DAG

那这里进行一个简单的建模

假设我有n条数据

有m个资源

共小n个module

那么假定

我们假定第i个

medule的吞吐为Oj

而调度的资源为Pi

大家可以很简单地看到

那么串行资源的延时

处理到完成

它的延时是这样的

而并行的最优状态

则是任意一级处理完

那么当然要满足如下的约束

也就是说

我们的管道一样粗

吞吐要一致

当这个约束满足的时候

实际上这两个公式是恒等

也就是说在理想情况下

它们完成的延时是一致的

并行DAG和串行DAG

都有它各自的优势

当然我们刚举的是理想情况下

事实上完全不是

就是限制时下的物理模型

远远比这个复杂

那么串行的模式的优势是什么

它整个模型是非常简单的

就是stagebystage

它的吞吐是非常高的

那么它的劣势是什么呢

它的数据时效性

和对于数据倾斜情况下的

对系统的整体延时的伤害

所以串行模型

我们总结它是面向吞吐

坚固延时

那么并行DAG

它的优势是说数据时效性好

它相对来说

对倾斜是友好

但是它的建模非常复杂

调度也是非常复杂

大家要知道

并行模型

要使得每个结点的处理规模

也就是吞吐对齐

这个对调度的在线的

Runtime时候的资源的调度

要求是非常高

它是面向延时而坚固吞吐的

一种策略

最后我们也在提到

前面也提到

整个流计算

它的进程是个longlive的进程

所以业界之前的调度系统

针对任务结束后

进程回收的情况

很明显不再适用

那么离线里面

无论是Yarn

开源的Yarn还是伏羲

都不能适应长进程的任务调度

现在有一个开源项目叫Slider

它在某种程度上

尝试去解决这个问题

阿里巴巴

我们的计算团队

我们建立了一个最新的系统

去解决这个长进程的问题

那么长进程调度和资源隔离

包括超卖模型

不是本课程的一个重点

这里我重点提几个问题

在离线和在线混合部署的情况下

IO的QS

IO的隔离将会是一个非常非常

棘手的问题

那么长进程延时SSA

与CPU平均利用率之间的矛盾

我们需要去做tradeoff

那么在我们的系统里

为了tradeoff这两者

可以让用户去设置minCPU

maxCPU

也就是ShareCPU

什么意思

就是说我即便这个进程

没有数据处理

我有绑定两个核

不能被其他进程抢占

但是我最多可以抢占别人

30个核

达到我占用max32个核的目的

当然有抢占就有优先级

系统需要进程去设置优先级

所以大家可以看到

如果我们的min设得非常大

接近max

说白了

你这个进程是一个独立的进程

也就是说你的CPU

是不会被别人抢占的

这个对于你自己的服务质量

保证是最好的状态

但是你没有数据的话

计算也没有

也就是说你这个进程

你这个CPU是被你独占的

所以在这种情况下

CPU的平均利用率

是相对比较低的

当我的min和max

相差比较大的时候

当然你的CPU有可能被别人抢占

所以对你响应的延迟

是有一些影响

但是我们线上的评估

影响在10%以下

当然我们这个利用的Cgroup的一些特性

那么长进程和离线 短进程

在申请方式 部署方式

拉起方式和包管理上面

都有非常大的不同

因为在线的请求

它的延时本身就在毫秒级别

如果还用离线的包管理方式

就把包下发下来

几百毫秒甚至一秒就过去了

这个对在线的服务是不能忍受的

对于资源的约束也完全不一样

在线因为对于延时

有比较高的要求

所以它资源占用的方式

摆放的方式是极其复杂的

无论是机房

核心交换机Rack

以及Rack里头的路由组

甚至不同机器副本 租户等等

有非常复杂的摆放策略

所以这就约束了我们的调度系统

一定要支持灵活的tag组合

另外

在（）的状态下恢复

资源不对齐的情况

都对长进程调度

提出了非常高的要求

在线系统如何接入这样子的调度

它的改造成本

也是我们考量的一个核心问题

总之

在线系统的调度

与离线系统的调度

差异性是非常大的

都是我们需要面临的解决的问题

在隔离纬度上

用户使用的Memory

网络 CPU

当然最难是Memory

和netWork的隔离

那么本地IO的访问 隔离

也是比较困难的

那么框架使用的资源

可以通过消息来进行软的流控

来限制

而Memory可以通过不同的JVM可以通过

不同的MemoryPut设置

那么NetWork可以通过

iptable+tc模块

当然这些都是我们目前

也不断一直在做的

包括最近现在比较火的SDN

但是CPU相对容易一些

因为是系统内核提供了

这样子的抢占的模式

我们总结一下

整个流计算的Failover容错机制

那么Batch是容错的最小单位

是数据跟踪的最小单位

是输入输出的最小单位

是控制的最小单位

那么整个容错分为源头重建

和节点重建两种

那么全量输出和增量输出

无外部互相依赖

跟踪消息与消息体量级

离线跟踪 流式跟踪

在线跟踪

完全在实现方法上

策略上不一样

那么有状态计算

这个计算的Failover

它的checkpoint

它的内存重建

当然大家在这里可以去看

关注开源的tachyon

那么在整个Failover的

机制设计方面

有运行时效率

和恢复时效率的一个tradeoff

包括如何避免雪崩

这些都是在容错机制上

要考虑的重点问题

综上所述

我们可以看到整个系统

是在不断做TradeOff

使吞吐与响应时间的TradeOff

即是实时性

与数据链的不可控的TradeOff

是非幂等操作

与链路的不可控的TradeOff

是精度与成本的TradeOff

是恢复成本

与运行时成本的TradeOff

是全链路与系统边界的Tradeoff

是需求多样性

与平台一致性的TradeOff

是不同计算场景

不同技术体系的TradeOff