1. 核心思想

通过把状态变成流的一部分，引入循环图，使得上一个item的状态可以作为下一个item的输入，使得有状态算子变成无状态算子。
$B(x,h)=Y \Longrightarrow B(x,s_t)=(y,s_{t+1})$
$x，Y$ be input and output items, $h$ ,the state handler and $s_t$ , the state object at time t.B becomes stateless and state management is done on the system side. This change allows the system to implement fault tolerance mechanisms, but it also opens the opportunity to implement deterministic processing

将状态当做流来处理

2. 计算逻辑图

Map applies a user-defined function to the payload of an input item and returns
a (possibly empty) sequence of data items with transformed payloads.
Broadcast replicates an input item to the specified number of operations.
Merge operation is initialized with the specified number of input nodes. It sends
all incoming data to the output.
Grouping constructs a single item containing a set of consecutive items that
have the same value of partition function. The maximum number of items
that can be grouped is specified as a parameter WindowSize.
The output item of the grouping has the same ordering label as the last
item in the output group. Groupings of different partitions are independent.
Grouping is the only operation that has a state.

MapReduce的逻辑执行图.png

在输入前front加入meta信息用来维护全局的顺序，item=(Payload, Meta)。
barrier用来过滤grouping算子由于输入元组乱序导致的错误输出。

The first map operation outputs mapped items according to map stage of
MapReduce model.
The grouping with WindowSize = 2 groups the accumulator with next
mapped item.
The combine map produces a new state of accumulator to be sent to grouping.
The final map converts accumulator into final reduce output.

reduce阶段的计算

排序规则保证每个accumulator item都在下一个没有被combined mapped item之前到达grouping。也就是说前一个item的状态和下个一item交替到达grouping。

3. 对乱序的处理：发送tomb使得乱序的输出无效

3.1无序产生的原因

每个operation在入口都会有一个用户定义的hash function（balance function）。由balance function决定将数据路由到对应的partition中，每个partition由单一的结点进行处理。对于reduce阶段，由于在map后的shuffle过程中，属于同一partition的数据来源于不同的上游物理节点，不能保证partition内的item是按序到来的。

3.2grouping opearation对于乱序的处理

只有grouping operation是有状态的，因此grouping对输入顺序敏感。当grouping接收到的items的时间小于已经被处理过的items,也就是说到来的的item是乱序时，则grouping算子将该乱序的item插入相应的位置，且对应顺序之前产生的item，则会相应的发送对应的tombstone，表示该item是无效的。tombstone的payload与无效item相同，仅在meta数据中，tombstone会被标记。这么做是为了保证tombstone在通过partition function时能被分配到相同的partition，从而保证与对应的由于乱序导致计算错误的item走相同的路径。

乱序的处理

如上图所示，grouping已经处理到item序号为6的元组，这时收到item为4的元组，则超前计算的item 4、5被视为无效，同时grouping先下游发送无效元组对应的tomb，之后发送正常的元组4。对于乱序产生的无效item，则会通过在下节介绍barrier过滤无效信息

4.使用barrier过滤

4.1全局顺序的定义

顺序定义
如图2所示，1'是1的衍生item，2'是2 的衍生item。即衍生item会被放在原始item之后。
meta用于维护全局顺序，在source节点处加入，组成<payloads,meta>的元组对，称作一个item。

使用acker跟踪

采用与storm类似的track机制跟踪元组，同一个item在发送和接受时会向acker发送相同的ackval，为了保证处理信息不丢失，算子在处理完成后先向acker发送send的ack，在发送receive的ack，如图6所示，当item到达barrier时，GT(global time)对应xor值是0。整个流处理系统正在处理中的item最小GT，即使acker中 XOR为非0值的最小GT.在图6中为21。