Heron 论文翻译及理解

背景介绍：

Heron是号称Twitter流数据处理的新一代实现，是StormV2。我们首先回顾一下Storm系统的问题

worker日志混乱，如果一个bolt日志过大，会冲掉其他bolt的日志
worker之间因为没有资源隔离，因此会出现不确定的worker间相互影响
Nimbus单点故障
背压(Back-Pressure)问题:receiver无法处理消息时，sender会丢弃消息。这种情况下如果没有ack将无法得知是否所有消息是否被处理，在最极端的情况下，系统会不同的消耗资源但得不到任何结果。

针对这些问题，我们实现了Heron：

数据模型：支持At Most once 以及At least once

工作模型：

0. 调度器（基于Mesos的Aurora）：

用于对每个topology使用的资源进行分配。

1. Topology Master （TM）

在调度器分配资源后，topology内部有一个类似选主的过程，主通过在zk上锁定一个公认的node以标示自己为主。同时，TopologyMaster也作为监控中心收集topology内部运行状态。但不作为数据流的中心，因此topology Master不会成为瓶颈。

2. Stream Manager（SM）以及Heron Instance（HI）

被分配到同一个topology中的SM之间建立全连接，即O（N*N）的连接，可见单个topology规模越大，SM建立的连接数量也会成倍增加。

HI则为分配到SM内部的处理单元，HI通过SM的信息通路进行通信。

3. Topology Back Pressure

背压机制主要用于流量控制，该机制保证了不同的组件可以以不同的速度处理事件。
举例而言：拓扑中有上游组件以及下游组件，当下游组件因为意外处理速度变慢时，上游组件依然以原速度发送事件，则会导致事件出现堆积。无论是中间丢弃事件还是把消息队列撑爆，显然都是我们不希望看到的情况。
这时候，背压机制的作用就是调节上游组件，让其减慢速度。

3.1 尝试1：TCP 背压

我们尝试在TCP连接的层面对该问题进行调整。SM之间通过TCP进行连接，二者之间都有Sending Buffer以及Receiver Buffer。当receiver处理速度变慢时，Receiver Buffer会随之变满，而后SendingBuffer也会慢，这时候上游组件是可以察觉到这种情况的。而后可以进行其他措施进行处理，如发送给其他的SM，或者调整自己的处理速度，直到下游SM速度赶上来位置。

这种方式实现简单，但实际效果并不理想。Because multiple logical channels (between HIs) are overlaid on top of the physical connections between SMs. 因为HI之间的多个逻辑连接都建立在SM的连接至上，该策略不仅降低了上游SM的发送速度，同时也降低了下游被背压的HI的处理速度。

其结果是，任何处理速度下降引起的TCP背压都会导致一连串的连锁反应导致系统处于长期的动荡调整阶段。

3.2 尝试2: Spout 背压：

这种方式同样结合了Sending Buffer以及Receiver Buffer进行。当SM意识到自己内部的某个HI处理速度变慢时，SM找到对应HI的spout并停止从该Spout中读取数据。而该Spout的发送缓存也因此会被填满并最终导致阻塞。与次同时SM会发送“开始背压消息”（Start BackPressure Message）到该topology的其他SM中。收到“开始背压消息”的SM则全部停止从自己上游获取消息的举措。
即：让其他所有的组件全部停下等待速度慢的组件消化完自己Receiver Buffer中的消息之后大家在一起继续。
当缓慢的HI速度赶上来之后，SM再次发送“停止背压消息”，收到消息的SM从新进入正常工作状态。

分析：该方式就是让其他所有快的都等慢的。同时隐含着系统被大量 stop start back pressure冲毁的风险。但其好处也是显而易见的，无论topology的层次有多深，该方式都能够保证足够低的处理延时。

3.3 尝试3：

层层背压：没懂，还好也没用

3.4实现：

我们最终实现的是Spout背压机制。该机制在实践中工作良好且可以轻易得出哪里是问题的root cause。
在实现上，每个socket连接都具有两个界限值，即高水位界限值以及低水位界限值。当receiver buffer到达高水位界限值时触发背压策略，直到Buffer回复到低水位时终止背压。设计成这种策略主要是防止系统反复震荡。
因为有了背压策略，所以系统不会在丢弃tuple，因此系统只有在出现组件故障时才回丢弃tuple，这使得tuple的丢弃更具有确定性。

当此时注意到，一旦进入背压模式，整个系统的处理速度与系统中最慢的组件速度相当。

4. HI 的实现

HI 抛弃Storm原有的多个Spout或者bolt可以共享一个JVM的思路，每个HI单独使用一个JVM，因此这样可以方便的调试各个组件的性能。

与此同时，因为传输事件的任务交由SM完成，因此对于我们而言，HI可以是以任意语言编写的代码。

对于HI的实现，我们主要有两种设计思路：单线程的实现以及多线程的实现。下面我们将分述两种实现。

4.1 单线程实现

在单线程的实现中，线程维护一个与其所属的SM的TCP连接。一旦有事件到达，users logic代码会被调用，如果user的代码会生成tuple，则该线程同样负责将该tuple传递给所属的SM。

   该代码虽然简单，但存在诸多问题，最主要原因是usercode可能会各种原因被block

（1）为了减速小睡一会
（2）调用系统IO，TCP连接，文件操作等需要上下文切换频繁的工作
（3）调用一些同步线程（synchronization）方法

这些block最关键的问题是会导致一些定时操作被耽误，例如性能状态监控。因为这些block时间不够确定，因此可能导致性能状态metric收集不及时，让master以为HI处于非良好状态。

4.2 双线程实现

如名所属，该方式下每个HI通过两个thread实现，一个GateWay Thread一个Task Execution Thread。

如图所示GateWay Thread负责与SM以及状态监控的MM进行通信，收发消息。Task Excurtion Thread 负责执行用户的逻辑代码。

两个线程之前通过三条单向有界队列（unidirection Queue）进行连接。当Data in 队列满时，GateWay不再从SM中读取消息，与此同时触发其所在SM的背压机制。同样，当dataout队列满时（由此可知队列buffer应该都由gate thread维护），GateWay同样认为此时TaskExcution线程不能接收更多的消息从而定制发送消息。与此同时TaskE线程也不再向dateout队列发送消息。

对TaskE线程而言，如果他是spout，则该线程不停地执行nextTuple方法并向data out队列推送消息。如果是bolt，则每个从datain 进入的消息都会触发 execute方法。

GC问题：
当我们使用topology的有界队列执行大型topology时，我们进场遇到GC问题。正当一切都正常的时候，一个网络中断（network outage）发生了，因此SM无法消费GateWay发出的消息因此dataout队列会发生堆积，且因为这些事件需要被处理，因此无法被GC回收。因此很容易导致HI达到其内存极限。
此时，当网络恢复时，Gateway从SM读入新的tuple的同时也向SM发送堆积在outqueue中的消息。如果GateWay读入消息在前，新object的生成占用更多内存从而触发GC，而之前outqueue因为堆积占用了所有内存，因此会导致更严重的性能下降(Performance degradation)。

为了解决此类GC问题，我们主要的思路是定期检查datain queue 以及dataout queue的大小，并动态更改其capacity。如果这些队列的capacity超过了某个设定的极限，则系统以折半的方式减小其capacity。
（这里没有说具体减小的方式，个人认为对于缩减datain queue，就是不在从SM中读取数据，知道datain queue中的tuple被消耗到一半之后完成一次折半capacity的过程。同样，对于out queue，也是先停in queue，然后直到out queue中消息被发出去一半之后再继续）。
这种折半减少capacity的方式被定时的调用直到：queue的capacity到达一个稳定的状况或者queue的大小减到0。如果queue的大小减到0，显然对于大多数的情况下，没有tuple可以被生产以及消费，（系统进入一个类似无状态的过程），此时进行GC会变得容易得多。

相反如果queue中的tuple数量小于设置值，系统会（通过GateWay从SM中读入更多的数据）增加queue中tuple的数量，知道到达极限值或者设置值。

6. MetricManager

MM负责收集并发送所有的状态。包括系统状态以及用户定义的的状态。

7. 开始过程以及异常处理

启动过程：
在提交一个topology之后调度器（Aurora）allocate必须的资源并在这些机器上启动topology container。Topology container中选出Topology Master，并在zk上通过锁住一个公认的节点宣布其为Master。与此同时，所有的SM（Stream Master）通过zk发现TM并与之进行定期心跳。
当所有SM与TM全部连接完毕(不需要两两互联)，TM开始运行一个部署算法将topology中不同的components分配到不同的container中。在我们的术语中，我们称之为physical plan。完成Physical Plan之后SM与TM通信获得其按照Plan应该与之互联的SM。此时，SM已经完成了一个完全互通的网络。HI启动并按照Plan获得其需要执行的任务，至此，整个topoloogy构建完毕，数据按照规定开始流动。
为了保险起见，TM可以Plan写入ZK以防止TM单点故障。

异常处理：

在Topology执行的过程中，有几种故障情况可能引起topology的整个执行过程。

TM Die：

当TM挂掉时，container会重启TM进程，TM进程则会自动从ZK回复状态。
（个人问题1： TM实时将状态写入ZK？包括哪些状态？）
如果存在Standby TM，则主备切换。与TM建立Channel的SM全部与之前的备建立Channel。

SM Die：

与TM类似，SM死掉时，它会被其所在的Container重启，被重启之后的SM首先联系ZK找TM，之后与TM通信下载Plan，以查看是否有任务变化。同时，与该死掉的SM互联的SM在与其失去连接后，同样连接TM下载新Plan查看是否应该找个新的SM连接还是等待其重启。

HI Die：

HI重启之后则直接与自己的SM连接获取plan并重新工作。

译者补充:

这里讲的比较模糊，有几个问题没有讲清楚：