实时计算与离线计算--前世今生

清疚 2024-04-28 13:05 62阅读 0赞

## 实时计算与 离线计算–前世今生 ##

### 离线计算 ###

#### MapReduce ####

Google的三篇论文开启了大数据处理的篇章，其中MapReduce被各大公司作为数据处理的主要方案。MapReduce的思想也是从早期的函数式编程语言中借鉴而来，推广到了分布式系统中，接触的东西多了，发现原来很多知识都是相通的。（很多初入IT行业的新人，面对不断出现的新技术往往会比较迷茫，到底该学哪一个呢，其实找到一个自己感兴趣的方向，并努力探索下去，相信慢慢就会一通百通，当然我也是个新银，个人观点，仅供参考^ ^）

传统的数据处理方式通常是将数据导入至专门的数据分析工具中，这样会面临两个问题：

1、如果源数据非常大时，往往**数据的移动**就要花费较长时间。

2、传统的数据处理工具往往是单机模型，面对海量数据时，**数据处理**的时间也是一个很大的问题。

**MR用直接在HDFS上运行防止了避免了大量数据移动，用分布式加速了数据处理**

**MapReduce是离线批量计算的代表，采用移动计算优于移动数据的理念，计算任务通常直接在HDFS的datanode上运行，这样避免了数据的移动**（当然reduce阶段还是需要节点间传输数据），并且采用并行计算的方式，大大减少了数据处理时间。

![img][]  
如图所示，MR模型将数据切分成大小相同的数据块，分别在对应的数据节点上启动map任务，map任务并行读取HDFS上的数据，根据reduce任务的个数将结果写入不同的临时文件中，map任务计算完毕之后，reduce任务从map节点处拉取自己负责处理的数据，并输出最终结果。

##### **wordcount例子：** #####

![img][img 1]

拿wordcount例子举例，map任务读取自身负责的数据，统计词频后，中间结果发送到对应的reduce节点，reduce节点负责统计最终的词频结果。

MR结合HDFS通过并行计算的方式，很大程度上解决了我们对数据存储、计算的需求，但是大家并不满足于现状， 慢慢的大家发现mr存在的一些问题;

##### MR存在一些问题 #####

###### 1.API较单一，对复杂的迭代计算并不友好 ######

> 下面我们来看一个解方程的例子：
> 
> ![img][img 2]

如图，上过初中的人一眼都能看出X=1，但是计算机不会因式分解呀，那如何利用计算机思想来求解方程呢？

首先，我们需要将方程转换（图中红色部分），将其看成一个级数，并且保证该级数收敛。

然后，在该级数收敛区间假设一个初始值，这里我们选取了X=1.5，代入方程右端，求出新的X值。

最后，将新的X值不断的进行迭代，当第K+1次和第K次的结果的差值小于我们预先定义好的阈值时，我们可以认为此时的X值已经无限接近于真实值，计算结束。

*对计算机解决数学计算感兴趣的同学可以复习一下线性代数和数值分析*

这里我们可以看到，计算机求解方程需要多次迭代，如果用MR实现，每一次迭代都要启动一次MR任务，这仅仅还只是一个最简单的例子，当面对数据挖掘这类涉及到比较复杂的组合计算时，采用MR效率就显得比较低下了。

###### **2.MR任务不能有效的利用内存** ######

![img][img 3]

map任务在处理数据后，会根据reduce的个数，生成对应个数中间文件，这些文件保存在磁盘上，map计算完成后reduce将文件fetch过来进行汇总。也就是说如果采用MR，**无论输入源数据有多少，哪怕内存放得下，中间结果还是需要落盘**，而我们知道io是计算机中最耗时的操作，这也是造成MR效率较低的一个原因。

##### MR的缺点： #####

###### 1、API复杂， ######

###### 2、不能有效利用内存 ######

**从上面的介绍可以看出MR存在两个缺点：**

1.无论源数据有多少，shuffle过程的中间数据都需要落盘，效率较低，不能很好的利用内存。

2.MapReduce只提供了map和reduce函数，api不够友好，编写代码时，首先都需要将计算思想转化成MapReduce模型，**非常反人类**（写过MR的应该非常有感触），而且稍微复杂点的计算需要多次迭代（尤其现在都在提人工智能，深度学习，这些都是比较复杂的逻辑计算），迭代启动任务也有较大的开销。

### Spark ###

#### Spark的job逻辑图 ####

Spark对这两点进行了优化，Spark也是一个批量计算框架，我们来看一下Spark的job逻辑图：

![img][img 4]

Spark的将数据抽象成RDD，RDD：**弹性分布式数据集（Resilient Distributed Datasets）**，这是一种分布式的内存抽象，允许在大型集群上执行基于内存的计算，与此同时还保持了MapReduce等数据流模型的容错特性。RDD可以常驻的内存的属性，大大简化了迭代计算所需的开销，Spark任务可以立马利用上一次计算出来的RDD来进行下次迭代。

Spark不仅仅提供Map和Reduce函数，还提供了额外的api，比如join，图中我们看到数据源data1和data2在中间处进行了join，丰富的api大大简化了编码逻辑，开发更高效（更详细api的可以在[Spark官网][Spark]了解）。

#### RDD优化 ####

需要注意的是内存不够用时RDD也可以采用磁盘作为其第二存储介质，而且如果计算中包含shuffle过程，为了保证容灾，shuffle处的RDD也是需要落盘的。

上图中我们可以看到任务被划分为3个stage，这里简单解释一下stage的意义，如果采用MR的运行方式，每个partition都启动一个task，stage1应该启动6个task，因为RDD需要存放在内存（内存不够时放磁盘），这样来看开销还是很大的，而且上下游task的RDD传输也会占用不小的开销。

Spark为了提高效率，采取了如下优化：当下游的RDD和上游的RDD是NarrowDependency时（也就是上游partition和下游partition是1对1，或n对1的关系，不涉及shuffle逻辑，比如map，flatmap等操作），将这些RDD放在一个task里面执行，大大减少了计算时所需的内存开销，也减少了网络传输的数据量，详细可以[参考][Link 1]。（很多计算框架都有采用这种优化，比如Flink）

#### 总结一下Spark的优点 ####

1.它不止提供了map和reduce逻辑，还提供了额外的api，更好的支持了复杂的迭代计算，如：groupby、join等函数。

2.更好的利用内存：内存够用的情况下，数据直接在内存进行处理，减少数据落盘次数，提高计算效率。

这时候问题来了：那既然Spark这么好，是不是Spark可以完全取代MR了？

答：也没有那么绝对，任何东西都有它存在的理由，都有它发光发热的地方，比如当数据量非常大，Spark内存放不下的时候，数据也是需要落盘的。对海数据简单清洗，排序时，Spark的性能并不一定就比MR好，所以还需要看具体的业务。

*而且前两年大家都在说Spark将会取代Hadoop，与其说是取代Hadoop，倒不如说Spark有可能取代MapReduce，因为Spark自身是没有存储功能的，数据源往往依赖HDFS存储。*

MR和Spark采用批量的方式解决离线计算业务，那想象一下另一些场景：在金融业务中，我们希望能够马上检测到有问题的账单，或者我们希望实时得到系统状态的统计信息，若采用批量框架，很难及时的得到反馈结果，这时就需要实时计算。

### 实时计算 ###

上篇文章大致介绍了离线计算MapReduce和Spark，但是无法满足对实时性要求较高的业务，下面我们来了解一下实时计算。

#### 离线和批量、实时和流式 ####

在聊实时计算之前，先说一下我对离线和批量、实时和流式的一些看法。

我们首先来简单看一下计算任务的大致流程：

![img][img 5]

**首先先说下批量计算和流式计算：**

图中显示了一个计算的基本流程，receiver处负责从数据源接收数据，并发送给下游的task，数据由task处理后由sink端输出。

以图为例，批量和流式处理数据粒度不一样，批量每次处理一定大小的数据块（输入一般采用文件系统），一个task处理完一个数据块之后，才将处理好的中间数据发送给下游。流式计算则是以record为单位，task在处理完一条记录之后，立马发送给下游。

假如我们是对一些固定大小的数据做统计，那么采用批量和流式效果基本相同，但是流式有一个好处就是可以实时得到计算中的结果，这对某些应用很有帮助，比如每1分钟统计一下请求server的request次数。

**那问题来了**，既然流式系统也可以做批量系统的事情，而且还提供了更多的功能，那为什么还需要批量系统呢？因为早期的流式系统并不成熟，存在如下问题：

1.流式系统的吞吐不如批量系统

2.流式系统无法提供精准的计算

*后面的介绍Storm、Spark streaming、Flink主要根据这两点来进行介绍。*

#### 批量和流式的区别 ####

##### 1.数据处理单位 #####

批量计算按数据块来处理数据，每一个task接收一定大小的数据块，比如MR，map任务在处理完一个完整的数据块后（比如128M），然后将中间数据发送给reduce任务。

流式计算的上游算子处理完一条数据后，会立马发送给下游算子，所以一条数据从进入流式系统到输出结果的时间间隔较短（当然有的流式系统为了保证吞吐，也会对数据做buffer）。

这样的结果就是：批量计算往往得等任务全部跑完之后才能得到结果，而流式计算则可以实时获取最新的计算结果。

##### 2.数据源 #####

批量计算通常处理的是有限数据（bound data），数据源一般采用文件系统，而流式计算通常处理无限数据（unbound data），一般采用消息队列作为数据源。

##### 3.任务类型 #####

批量计算中的每个任务都是短任务，任务在处理完其负责的数据后关闭，而流式计算往往是长任务，每个work一直运行，持续接受数据源传过来的数据。

##### 4.离线=批量？实时=流式？ #####

习惯上我们认为**离线和批量等价；实时和流式等价**，但其实这种观点并不完全正确。

假设一种情况：当我们拥有一个非常强大的硬件系统，可以毫秒级的处理Gb级别的数据，那么批量计算也可以毫秒级得到统计结果（当然这种情况非常极端，目前不可能），那我们还能说它是离线计算吗？

\*\*\*所以说离线和实时应该指的是：数据处理的延迟；批量和流式指的是：数据处理的方式。\**两者并没有必然的关系。事实上Spark streaming就是采用小批量（batch）的方式来实现实时计算。*

*可以参考下面链接：[https://www.oreilly.com/ideas/the-world-beyond-batch-streaming-101][https_www.oreilly.com_ideas_the-world-beyond-batch-streaming-101]。作者是Google实时计算的负责人，里面阐述了他对批量和实时的理解，并且作者认为批量计算只是流式计算的子集，一个设计良好的流式系统完全可以替代批量系统。本人也从中受到了很多启发。*

介绍完这些概念后，下面我们就来简单看看目前流行的实时计算框架的实现和区别。

### Storm ###

#### Storm介绍 ####

Storm采用ack机制实现了数据的重放,由此导致了吞吐不如批量计算，但这不意味着批量流计算的吞吐不如批量计算

Storm做为最早的一个实时计算框架，早期应用于各大互联网公司，这里我们依然使用work count举例：

![img][img 6]

spout：负责从数据源接收数据

bolt：负责数据处理，最下游的bolt负责数据输出

spout不断从数据源接收数据，然后按一定规则发送给下游的bolt进行计算，最下游的bolt将最终结果输出到外部系统中（这里假设输出到DB），这样我们在DB中就可以看到最新的数据统计结果。Storm每一层的算子都可以配置多个，这样保证的水平扩展性。因为往往处理的是unbound data，所以storm中的算子都是长任务。

容灾

容灾是所有系统都需要考虑的一个问题，考虑一下：假如运行过程中，一个算子（bolt）因某种原因挂了，Storm如何恢复这个任务呢？

![img][img 7]

批处理解决方案就比较简单，拿MR举例，假如一个运行中map或reduce失败，那么任务重新提交一遍就ok（只不过重头计算又要花费大量时间），下面我们看看Storm是如何解决的：

storm的spout有一个buffer，会缓存接收到的record，并且Storm还有一个acker（可以认为是一个特殊的bolt任务），每条record和该record所产生的所有tuple在处理完成后都会向对应的acker发送ack消息，当acker接收到该record所有的ack消息之后，便认为该record处理成功，并通知spout从buffer中将该record移除，若receiver没有在规定的时间内接收到ack，acker则通知spout重放数据。

acker个数可以由用户指定，因为数据量比较大时，一个acker可能处理不过来所有的ack信息，成为系统瓶颈（如果可以容忍数据丢失，当然也可以关闭ack机制，可以显著提高系统性能）。并且acker采用了巧妙的机制，优化了ack机制的资源占用（有兴趣的同学可以参考官网，网上也有很多博客介绍ack具体实现）。

Storm采用ack机制实现了数据的重放，尽管做了很多优化，但是毕竟每条record和它产生的tuple都需要ack，对吞吐还是有较大的影响，关闭ack的话，对于某些不允许丢数据的业务来说又是不可接受的。

Storm的这种特点会导致大家认为：**流式计算的吞吐不如批量计算**。（这点其实是不对的，只能说Storm的设计导致了它的吞吐不如批量计算，一个设计优秀的流式系统是有可能拥有和批处理系统一样的吞吐）

#### **数据不重不丢** ####

之前我们提到早期的流式系统无法提供精准的计算服务，下面我们详细了解一下：

![img][img 8]

sink处的重复输出：假如运行过程中，boltA数据入库后，boltB因为某种原因crash了，这时候会导致该record重放，boltA中已经处理过的数据会再次入库，导致部分数据重复输出。

不仅sink处存在重复输出的问题，receiver处也同样存在这种问题。（在讲解Spark streaming处会详细介绍什么情况下receiver会重复接收数据）

Storm没有提供exactly once的功能，并且开启ack功能后又会严重影响吞吐，**所以会给大家一种印象**：**流式系统只适合吞吐相对较小的、低延迟不精确的计算；而精确的计算则需要由批处理系统来完成**，所以出现了Lambda架构，该架构由Storm的创始人提出，简单的理解就是同时运行两个系统：一个流式，一个批量，用批量计算的精确性来弥补流式计算的不足，但是这个架构存在一个问题就是需要同时维护两套系统，代价比较大。

那么有没有一种架构，可以满足高吞吐、低延迟的要求，同时也提供exactly once功能？有的，下面我们来看看Spark streaming

### Spark streaming ###

*但也因为处理数据的粒度变大，导致Spark streaming的数据延时不如Storm，Spark streaming是秒级返回结果（与设置的batch间隔有关），Storm则是毫秒级。*

#### 吞吐 ####

Spark streaming采用小批量的方式，提高了吞吐性能：

![img][img 9]

这里我们简单展示Spark streaming的运行机制，主要是与Storm做下对比。Spark streaming批量读取数据源中的数据，然后把每个batch转化成内部的RDD。**Spark streaming以batch为单位进行计算（默认1s产生一个batch），而不是以record为单位**，大大减少了ack所需的开销，显著提高了吞吐。

> *但也因为处理数据的粒度变大，导致Spark streaming的数据延时不如Storm，Spark streaming是秒级返回结果（与设置的batch间隔有关），Storm则是毫秒级。*

#### 不重不丢（exactly once） ####

Spark streaming通过batch的方式提高了吞吐，但是同样存在上面所说的数据丢失和重复的问题。

在解答这个问题之前，我们先来了解一下一些概念：

1.at most once：最多消费一次，会存在数据丢失

2.at least once：最少消费一次，保证数据不丢，但是有可能重复消费

3.exactly once：精确一次，无论何种情况下，数据都只会消费一次，这是我们最希望看到的结果

大部分流式系统都提供了at most once和at least once功能，但不是所有系统都能提供exactly once。

我们先看看Spark streaming的at least once是如何实现的，Spark streaming的每个batch可以看做是一个Spark任务，receiver会先将数据写入WAL，保证receiver宕机时，从数据源获取的数据能够从日志中恢复（注意这里，早期的Spark streaming的receiver存在重复接收数据的情况），并且依赖RDD实现内部的exactly once（可以简单的理解采用批量计算的方式来实现）。**RDD：Resilient Distributed Dataset弹性分布式数据集**，Spark保存着RDD之间的依赖关系，保证RDD计算失败时，可以通过上游RDD进行重新计算（RDD如何实现容错这里就不解释了，可以自行查资料）。

上面简单解释了Spark streaming依赖源数据写WAL和自身RDD机制提供了容灾功能，保证at least once，但是依然无法保证exactly once，在回答这个问题前，我们再来看一下，什么情况Spark streaming的数据会重复计算。

![img][img 10]

这里我们主要关注图中的3个红框：

Spark streaming的RDD机制只能保证内部计算exactly once（图中的1），但这是不够的，回想一下刚才Storm的例子，假如某个batch中，sink处一部分数据已经入库，这时候某个sink节点宕机，导致该节点处理的数据重复输出（图中的3，Storm处已经解释过了）。还有另一种情况就是receiver处重复接收数据（图中的2），我们看一下receiver重复接收数据的情况：

![img][img 11]

假如receiverA目前从kafka读到pos=100的记录，并且已经持久化到HDFS，但是由于网络延迟没有及时更新pos，此时receiverA宕机了，receiverB接管A的数据，并且后续的任务还会从pos=100处重新读取，导致重复消费。造成这种情况的主要原因就是：**receiver处数据消费和Kafka中position的更新没有做到原子性**。

根据上面的讨论，可以得出：

#### 一个流式系统如果要做到exactly once，必须满足3点： ####

1.receiver处保证exactly once

2.流式系统自身保证exactly once

3.sink处保证exactly once

这里数据源采用Kafka举例是因为Kafka作为目前主流的分布式消息队列，比较有代表性。Kafka consumer的position可以保存在ZK或者Kafka中，也可以由consumer自己来保存。前者的话就可能存在数据消费和position更新不一致的问题（因为无法保证原子性，也是之前Spark streaming采用的方式），而采用后者的话，consumer可以采用事务更新的方式（写本地或者采用事务的方式写数据库），保证数据消费和position更新的原子性，从而实现exactly once（[参考][Link 2]）。

#### Spark streaming 实现 exactly once ####

Spark streaming1.3版本新添加了[Kafka Direct API][]来实现数据接收的exactly once，本质上就是上面提到的后者，Spark streaming自己维护position，streaming的worker直接从Kafka读取数据，position由Spark streaming管理，不再依赖ZK保存，同时保证数据消费和更新position的原子性，从而实现exactly once。

*并且新的方式已经不再需要receiver持久化数据，因为Kafka本身就支持数据持久化，可以避免receiver处持久化数据的开销，实现exactly once的同时也提高了性能。*

而sink处的exactly once的实现则视外部系统而定，比如文件系统本身就支持**幂等**（同一个操作执行多次，不会改变之前的结果），同时Spark streaming也提供了api，用户可以自己实现sink处的[事务更新][Kafka Direct API]，receiver、sink和Spark streaming三者结合起来才能实现了真正的exactly once。

*Storm trident本质上也是采用了小批量的方式，并且也实现了exactly once语义，这里就不做过多讨论。*

> \*直到这里，我们了解到Spark streaming拥有较好的吞吐和exactly once语义，解决了Storm一些不足，\*\*是不是只有采用类似Spark streaming这种小批量（micro-batch）的方式才能实现这些功能？\**答案是：NO。下面我们来看看Flink。*

### Flink ###

Flink在数据处理的方式上和Storm类似，并没有采用小批量，是一个真正的流式系统。它不仅拥有了不弱于Spark streaming的吞吐，并且提供了exactly once语义。既然Flink也是逐条处理记录，那么它是怎么做到的呢？跟上我的脚步…（下面内容大部分参考官网，捡重点的翻译，想起来一个段子：如何快速成为业界大牛？答：翻译英文文档。。hahaha，开个玩笑^ ^）

简单来说，\***Flink采用轻量级分布式快照实现容错\***，大致流程是：Flink不断的对整个系统做snapshot，snapshot数据可以放在master上或外部系统（如HDFS），假如发生故障时，Flink停止整个数据流，并选出最近完成的snapshot，将整个数据流恢复到该snapshot那个时间点，snapshot本身比较轻量，而且用户可以自行配置snapshot的间隔，snapshot的性能开销对系统的影响很小（官方测试snapshot开启前后的性能差距不大）。

![img][img 12]

barrier是分布式snapshot实现中一个非常核心的元素，barrier和records一起在流式系统中传输，barrier是当前snapshot和下一个snapshot的分界点，它携带了当前snapshot的id，假设目前在做snapshot N，算子在发送barrier N之前，都会对当前的状态做checkpoint（checkpoint数据可以保存在外部系统中，如HDFS），checkpoint只包含了barrier N之前的数据状态，不会涉及barrier N之后的数据。

![img][img 13]

因为算子很多情况下需要接收多个算子的数据（shuffle操作），所以只有当所有上游的发送的barrier N都到达之后，算子才会将barrier N发送给下游（所有的下游）。当所有的sink算子都接收到barrier N之后，才会认为该snapshot N成功完成。

为了保证一致性，需要遵守以下几个原则：

1.一旦算子接收到某一个上游算子的barrier之后，它不能再处理该上游后面的数据，只有当它所有上游算子的barrier都到达，并将barrier发送给下游之后，才能继续处理数据，否则的话会造成snapshot N和N+1的数据重叠。

2.某个上游算子的barrier到达之后，该上游算子后续的数据将会被缓存在input buffer中。

3.一旦所有上游的算子的barrier都到达之后，该算子将数据和barrier发送给下游。

4.发送成功之后，该算子继续处理input buffer中的数据，并继续接收处理上游算子发送过来的数据。（有点啰嗦啊）

下面我们来看一个完整的snapshot流程图：

![img][img 14]

图片有点不清晰，可以自己去官网看，笔者比较蠢，怎么都截不下来高清的

图中的Master保存了snapshot的状态，假设数据还是从Kafka中获取，首先receiver算子会先将当前的position发送给master，记录在snapshot中，并同时向下游发送barrier，下游的算子接收到barrier后，发起checkpoint操作，将当前的状态记录在外部系统中，并更新Master中snapshot状态，最后当所有的sink算子都接收到barrier之后，更新snapshot中的状态，此时认为该snapshot完成。

通过这种轻量级的分布式snapshot方式，Flink实现了exactly once，同时Flink也支持at least once，也就是算子不阻塞barrier已经到达的上游算子的数据（多个上游算子的情况），这样可以降低延迟，但是不保证exactly once。

从图中我们可以看出Kafka position也是由Flink自己维护的，所以能够保证receiver处的exactly once，sink处也同样存在Spark streaming一样的问题，exactly once依赖外部系统或需要用户自己实现。Flink官网给出了目前支持的[Data Sources和Sinks][Data Sources_Sinks]以及容错的粒度。

> FLINK以为自己维护了快照的checkpoint 所以可以保证exactly once，并且因为snapshot实现了容错

*其中sink处采用Kafka的话不支持exactly once，个人猜想是不是因为早期的Kafka producer没有支持exactly once语义，而导致Flink无法支持。Kafka0.11版本中添加了producer exactly once的支持，是否后续能够添加进来？*

讲到这里，我们可以了解到：

1.流式系统并不一定就是吞吐差的代名词

2.流式系统也可以做到exactly once

就如Google流式系统负责人Tyler Akidau所说：**一个设计良好的流式系统是能够在吞吐完全媲美批量系统，并且提供精准的实时服务。***（那是不是以后可以完全用流式系统取代批量系统？）*

window和event time

Flink相比Spark streaming不仅提供了更低的延迟，而且Flink还对window和event time提供了更好的支持。window和event time又是什么呢？

**window**

现实生活中，大部分数据源其实是unbound data，没有边界，我们没有办法得到一个最终的统计结果，很多情况下我们会对固定时间间隔的数据进行统计，比如每5s统计一下服务器的qps，window机制能够帮我们很好的完成这项需求。

![img][img 15]

如图（标号代表事件发生的时间），流式系统会每隔5s创建一个window，将该时间段的数据放入buffer，累加后输出结果。图中0-5s产生的数据放在第一个window中（3s处有两条数据），累加后输出count=6。

*window类型也有很多种，上图是一个Tumbling Windows的例子，另外还有Sliding Windows和session window，具体区别读者可以自行查资料。*

上图是一个比较理想的示例图，理想很丰满，现实很骨感，事情往往不尽如人意（情不自禁的都想唱起来了：人生已经如此的艰难，有些事情就不要ao…流式系统的破事怎么这么多！！），直接按接收时间来划分window可能会存在误差：

![img][img 16]

假设由于网络延迟，应该属于第一个窗口的数据3延迟到达，被分到了第二个窗口，这时候计算结果并不准确。怎么办呢？

event time和process time：

假设一个流式系统目前正在接收并处理用户手机的日志，但是由于网络延迟，或者用户手机离线，导致日志没有及时发送到流式系统，流式系统观察到数据的时间和数据真正产生的时间可能存在偏差，我们把数据真正产生的时间叫做：event time，把流式系统处理该数据的时间叫做：process time。

![img][img 17]

event time和process time往往会存在延迟，这种不一致会导致数据乱序，如图所示：蓝色事件晚于黄色事件发生，但是事件的处理却先于黄色事件。

早期的流式系统并没有区分process time和event time，往往将process time等同于event time。针对这一问题，一个很直观的解决方案就是：让数据自身携带timestamp，该timestamp记录该数据产生的时间，即为event time，流式系统按数据的event time来将数据分配到对应的窗口，而不是按处理数据的时间。

window需要知道该窗口的数据都已经全部到达，然后触发计算逻辑，如何window判断时间T之前的数据是否都已经到达呢？

watermark

那就是引入watermark机制，watermark同样也携带一个时间戳，当算子接收到watermark T后，就代表时间T之前的数据已经接收完毕，不会再有小于时间T的数据。

![img][img 18]

如图：W(17）到达后，表示后续数据的时间戳不会小于17。那可能有人会问了：**那就是有一部分小于17的数据他喵的就是比w(17)还晚到了怎么办？**

watermark还会配合一个allow lateness参数，window接收到watermark后，再等待一段时间才会关闭窗口，如果这段时间有些数据依然没有发送过来，那就只能忽略它们了（window的内心os：我也尝试过等待，但我还有更重要的事情要做），而且考虑到流式系统的实时性，假如可接受的时间内，数据没有传输过来，那就算等到它过来再计算，从实时性这个角度来说，这时计算的结果也有可能也已经没有意义了。

Flink对window和watermark都提供了较好的支持，Spark streaming从2.0中也开始引入watermark功能，但是支持的功能有限，并且真正的流式可以更优雅、简单的实现window和watermark，从这个角度来看，Flink是优于Spark streaming的。

总结：

了解了Storm、Spark streaming、Flink各自的特点后，我们知道Storm提供了低延迟的计算，但是吞吐较低，并且无法保证exactly once（Storm trident采用batch的方式改善了这两点），Spark streaming通过小批量的方式保证了吞吐的情况下，同时提供了exactly once语义，但是实时性不如Storm，而且由于采用micro-batch的方式，对window和event time的支持比较有限（Spark streaming2.0中引入了window和event time，还在起步阶段）。Flink采用分布式快照的方式实现了一个高吞吐、低延迟、支持exactly once的流式系统，流式处理的方式也能更优雅的支持window和event time。

当然也不是说Flink一定就比Storm、Spark streaming好，没有最好的框架，只有最合适的框架，根据自身的业务、公司的技术储备选择最合适的框架才是正确的选择。

来源:简书

地址: https://www.jianshu.com/p/16323566f3c6

不如Storm，而且由于采用micro-batch的方式，对window和event time的支持比较有限（Spark streaming2.0中引入了window和event time，还在起步阶段）。Flink采用分布式快照的方式实现了一个高吞吐、低延迟、支持exactly once的流式系统，流式处理的方式也能更优雅的支持window和event time。

来源:简书

地址: https://www.jianshu.com/p/16323566f3c6

[img]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/28/50bf24d4c23e43928c3d3dc5f8c422d4.png
[img 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/28/141643850de54d3f955d1c01da203c25.png
[img 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/28/b62e300c918c495495e44c1cf1c145d9.png
[img 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/28/64a36037615f4f9e9b62a9f3940b17e1.png
[img 4]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/28/a208233abe5c43cbb2fdb312e39a8e8a.png
[Spark]: https://link.zhihu.com/?target=https%3A//link.jianshu.com/%3Ft%3Dhttps%3A//spark.apache.org/docs/latest/rdd-programming-guide.html%23transformations
[Link 1]: https://link.zhihu.com/?target=https%3A//link.jianshu.com/%3Ft%3Dhttps%3A//github.com/JerryLead/SparkInternals/blob/master/markdown/3-JobPhysicalPlan.md
[img 5]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/28/3a7bc5f2e7c14e348cd598adeaed9b97.png
[https_www.oreilly.com_ideas_the-world-beyond-batch-streaming-101]: https://link.zhihu.com/?target=https%3A//www.oreilly.com/ideas/the-world-beyond-batch-streaming-101
[img 6]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/28/3e6b211a9d51482b8d0ea2b455765ef5.png
[img 7]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/28/a5916ed0a6d34b1791094e1119418041.png
[img 8]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/28/2ab0b755dc64477b85bc0ed98c6fa84e.png
[img 9]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/28/aa03f87b74994a9b96c23d07cfd00676.png
[img 10]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/28/8a84269b000741c4b33b56eb0b5b9c1b.png
[img 11]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/28/b61a50fb7fc64776b9e36b94e6853707.png
[Link 2]: https://link.zhihu.com/?target=https%3A//link.jianshu.com/%3Ft%3Dhttps%3A//kafka.apache.org/documentation/%23semantics
[Kafka Direct API]: https://link.zhihu.com/?target=https%3A//link.jianshu.com/%3Ft%3Dhttps%3A//spark.apache.org/docs/latest/streaming-programming-guide.html%23fault-tolerance-semantics
[img 12]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/28/e97cdd8fecea439b9af953dedf2fac03.png
[img 13]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/28/d6a43786af544043b1ae8f3f787db6cc.png
[img 14]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/28/490520a18c7a46c7a5193c144a46300f.png
[Data Sources_Sinks]: https://link.zhihu.com/?target=https%3A//link.jianshu.com/%3Ft%3Dhttps%3A//ci.apache.org/projects/flink/flink-docs-release-1.3/dev/connectors/guarantees.html
[img 15]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/28/404dce9e76124bb19f96ae457aefd9bf.png
[img 16]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/28/95ee70645dad4fbba5105a7bb3fc77c6.png
[img 17]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/28/133ac7b170fa437cbfc0878734022f94.png
[img 18]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/28/de56d23d77714ac791f3bea28acf530a.png