业界典型系统技术概要分析（主讲人：强琦）在线视频

在流式计算领域

目前已经涌现出来了一些系统

目前的开源

已经有非常多成熟的实现了

我们下面

将会介绍几个经典的系统

下面我就会介绍业界比较经典的

几个流计算产品

Storm大家都比较清楚

它是Twitter内部使用开源

被广泛使用的一套流计算系统

那么它的一个核心概念是说

一个任务要创建一个Topology

它表示了一个完整的流计算作业

那么它的最开始的源头

名字叫做Spout

它是做收集数据的任务

它的前面

就可以挂任何的数据源

比如说卡夫卡

任何一个队列系统

甚至可以对接文件

那么Bolt是它的具体计算任务

所在的载体

而Bolt里头有诸多的Task

它是在Spout和Bolt里头

负责具体一个分片的一个实体

它也是Storm里头

调度的最小单位

那么Acker它是负责

刚才我们提到了

在流计算里头

整个的监控是在监控数据

而不是在进程的心跳

所以Acker是负责整个消息

跟踪机制

它是用来判断是否这些数据

被哪些结点处理

或者这个数据

经过了一个复杂的Topology

或者DAG

它最终被处理完了

而Storm的整个的容错

它是采用源头重发的消息机制

源头重发

在网络流量激增的情况下

会造成系统的雪崩风险大大提升

那么右图

是两个Storm的作业

它先从源头读出数据

然后进行filter 过滤

然后最终进行join

最后join后做进行一些逻辑处理

Storm采用了

Nimbus Supervisor

之间的方式进行任务调度和跟踪

它们之间是利用Zookeeper

来进行通讯

那么Nimbus

相当于一个全局的任务的Master

当然

它现在是一个全局的一个单点

现在是无状态

它负责接收Topology

然后进行二重的资源调度

并且将调度的信息

记录到Zookeeper中

定期检查Zookeeper中的各种

Supervisor的心跳信息

根据心跳状态

决定这个任务是否进行重新调度

而Supervsor的角色

充当着每台物理机的一个

watchdog

它也是一个无状态的角色

那么它在轮询Zookeeper中的

调度任务信息

然后接收到发现有

启动任务的信息

它就会拉启进程

启动Task

同时定期要把心跳信息写入

Zookeeper

以便Supervisor来做出重新调度

或者系统的重发的操作

Storm里头比较核心的一个

属于它的消息跟踪机制

大家都知道

有状态计算下

在流计算场景下

其实要做到准确

其实是非常非常困难

因为它是个有状态的计算

那么所有的消息

你要做到only once

也就是说当且仅当

处理一次这样的语意

其实是非常困难

大家又知道一个DAG

可能会出现一个结点

分裂出多个结点

一条数据经过UDTF

会分裂出多条数据

而多条数据又会重新地unit

在一起

或者join在一起

所以要判断一条数据

在整个DAG中执行流

要判断它是否被所有的结点

处理成功

这个还是相对比较困难的

而Storm设计的消息跟踪机制

来追踪源头信息的

所有的子孙的信息

那它的基本思路是说

Acker结点

是进行消息跟踪的结点

以源头的ID为hash key

来确定跟踪的Acker

源头信息对应的所有的子孙消息

都有跟Acker的负责跟踪

而消息树上

每产生一个新的子孙消息

则对应的对有地去通知到

对应的Acker

子孙消息被处理

然后再去通知到对应的Acker

当Acker里所有的子孙消息

都被处理的时候

那么整个数据处理

就完成了

子孙的产生是有父结点

而处理是被子结点

所以Storm在这里头

用了一个非常巧妙的

一个异或的一个方法

大家都知道

异或是可以满足交换的

也就是说成对出现的两个值

只要成对出现

无法保证它一定是连贯地出现

那么只要它是偶数出现

它就一定能保证

它这个异或的最终值可以变成0

Storm在利用

当父结点产生这个消息的时候

产生一个随机数

把这个随机数异或到Acker里去

Acker把这个随机数

传递到下一步的结点

当这个结点正确被处理以后

再把这结点发送给Acker去做异或

所以Storm就巧妙利用了

这个Acker机制

来压缩我们整个数据的跟踪机制

最终保证任意结点出现宕机

则这个值不会变成0

光有以上的机制

还远远达不到

消息的only once的语意

那么Storm为了解决这个问题

在0.7.0

当然trident后面的版本

改进了它的实现

它为了解决消息

被重复处理的问题

而源头数据重发的消息

又没有任何附加信息

用户也是没办法判断是否重发的

所以Storm利用将源头

串行划分为Batch

每个Batch赋予一个ID

利用Acker来判断Batch

是否处理完成

当有任何异常

包括超时的时候

Spout会重发

Batch内的所有消息

那影响中间状态的操作

是可以并行运行

那么用户的唯一代码

可以利用BatchID去进行去重

如果

当然Storm的Transaction机制

保证BatchID的严格递增

所以用户可以利用数据的版本

也就是说利用这个BatchID

来进行去重

如果这个ID比如说3

其实用户的State里头

里面已经有3

这个Batch ID的处理结果

那么这一次就可以把

bypass掉Batch ID为3的重发

这里有一个极大的限制

就是我刚才提到了

整个Topolgy同一时刻

能保证有状态的结点

进行操作的时候

它的Batch ID是串行序列化

保证在并发情况下

满足Transactional的语意

那么可以看到

Storm整个消息在框架里

它是不落地的

也就是说它不落盘

它是用Push的模式

上游直接Push给下游

它的数据处理是比较高效的

它保证了数据的不丢不重

Only once的语意

而且在Transaction的语意下

可以为各种消重提供了可能性

它的调度模式是非常非常简单

当然用户就可以去prading

你自己复杂的调度

Storm的社区也比较活跃

现在被很多公司采用

当然Storm也有它自己的劣势

就是我们刚刚说了

保证State在Batch ID下的一个

串行化的语意

会造成整个系统的吞吐会下降

而且它的源头重发

在网络抖动的情况下

极易造成雪崩

因为少数结点的超时宕机

引起整个Topology的重做

而整个Topology的重做

继而引发了网络更加的拥堵

另外Batch ID大小的选择

对于用户来说

根据业务划分

门槛比较高

用户设置这些参数比较困难

接下来

我们去介绍一下

Kinesis系统

Kinesis是（09：03）

开放的一种完全托管的

实时大规模数据流开放服务

它的特点

相对Storm来说

它是采用了结点内部重放的系统

而不是像Storm那样子源头重发

它的所有的结点

都已经在EC2中

这样它提供了比较好的调度策略

复用安全 资源隔离

借用EC2的良好的系统管控

它可以做到非常好的扩展性

弹性可伸缩

大家可以看到

Kinesis的计费

完全是在EC2可以兼容掉

它只支持单机的Task

当然我们可以用多机的虚拟

组成一个非常复杂的DAG-Task

这里用户也需要实现

自己Task内部的消重逻辑

来做到only once的语意

它的数据收集和计算

类似像Storm也进行了独立

而数据收集模块

对数据进行持续化

最长周期保留24小时

当然你可以以get方式

从其他系统获得Shard数据

而计算模块

处理被推送的数据

Instance的个数和整个数据

保留的Shard的个数相同

用户代码可以自主控制

有状态的数据的Checkpoint节奏

用户可以自主调用相应的接口

来调整并发度

那么每个Shard串行

将接收到的消息写到S3文件中

SplitShard后

原有Shard不再接受新数据

原有Shard

对应的App的Instance处理完后

消息关闭

再启用新的Shard

和对应新的Instance

这样子

它可以做到动态的并发地调整

那么这一条

对于我们其实刚开始说到的

实时数据的整个的规模的突发性

和离线完全不同

那么进行动态计算规模的调整

扩容 缩容

就显得尤为重要

那么它的计算可以更加的弹性

那么服务的可用性也更高

大家可以从示意图上看到

整个扩容的过程

接下来我简单地再介绍一下

MillWheel的系统

MillWheel是利用内部

支持Snapshot功能

然后Bigtable来进行

持续化中间结果

每个结点类似于Kinesis

每个结点的计算输出消息

都进行持久化

那么每一级消息

都去实现自己的不丢不重

也就是only once的语意

后面这两个系统

区别于Storm的时候

它没有复杂的跟踪树

因为每一级都把它的输出消息

进行持久化

用户可以通过Settimer

PrecessTimer接口

解决用户代码

在消息来的时候

才取得控制流的弊端

然后在源头消息

将数据打上时间戳

在每个内部的计算结点

输出所有Pipe上的最小时间戳

像所有输出Pipe

广播当前最小的时间戳

那么以方便用户利用这一机制

解决消息乱序

解决持续一致性的问题