2019年天猫双11再次刷新世界记录，订单创新峰值达到54.4万笔/秒，单日数据处理量达到970PB；支撑起这个双十一狂欢神话是阿里强悍的数据云上系统，这其中阿里的数据计算云引擎Blink功不可没，而Blink正是由Apache的顶级项目Flink经过"阿里化“而来；

四代计算引擎：

• MapReduce：批处理引擎，为其他计算引擎提供量大数据处理的核心思想：Mapper、Reduce；
• Storm：流计算引擎，为了满足更高的实效性而生的；
• Spark：集流式处理和批处理于一身的统一计算引擎，基于内存计算，提高性能(以批处理模拟了流式处理)
• Flink：另一款集流式处理和批处理于一身的统一计算引擎，吸收了以上三款引擎的优点进行改进后而来的计算框架(以流式计算模拟了批处理)

由此看出，计算引擎由批处理向流式处理发展，由一个组件干一个事向一个组件干全部事转变，一代代计算引擎的发展，追求的目标都是：更通用，更高效，更优雅，更易用的一栈式(集成批处理和流式处理)的计算引擎；

相较于Flink前三代计算引擎的区别与缺点：

• MapReduce作为初代计算引擎最大的价值在于提供了一种分布式计算的思想，但是随着业务中海量数据集越来越普遍，使用无限数据流处理而设计的系统处理数据场景越来越多，MapReduce效能和开发复杂度的局限性也突显出来；

• Storm是第一款流失计算引擎

  • 使用Clojure(基于JVM)语言开发，想看源码不方便；
  • Storm本身还存在一些bug，生产上发生过集群崩溃的事件；
  • 不具备一些流处理中需要的高级功能：例如exactly one等；

• Spark与Flink功能类似，Flink拥有Spark全部的功能，而Spark是当下最流行最完美的一套计算引擎，但是：

  • 核心理念不同：
    • Spark核心模块SparkCore是用于批处理的，批处理为常态，SparkStreaming是利用微批处理来模拟实现的流处理；
    • Flink的核心为流式处理，流处理为常态；利用流处理去模拟出批处理；Flink把核心放在流处理上，实现来一些高级也是必须的功能，如数据状态，事件时间，分布式快照，watermark等；
    • Flink支持增量迭代，具有对迭代自动优化对功能，在迭代数据处理上，比Spark更突出
    • Flink相较于Spark有更高对吞吐量，更强对计算性能，更低对延迟，更好对容错机制

Flink

Flink等诞生：

Flink源于柏林工业大学等一个研究性项目----StratoSphere；早期Flink是做批处理等，但是在2014年，StratoSphere里面的核心成员孵化出Flink，同年Flink捐给来Apache，后来成为Apache的顶级大数据项目，同时Flink将计算的主流方向定位为流处理；

Flink是什么：

Flink是一个低延迟，高吞吐、统一的大数据计算引擎；---- Stateful Computations over Data Streams，有状态的流式计算引擎。

Flink最区别于其他流计算引擎对就是statefule；即有状态计算 Flink提供流内置对对状态对一致性对处理，即如果任务发生流Failover，其状态不会丢失，不会被多算少算，同时提供流非常高对性能；

g8kh78syvr Flink01

• Streaming-first
• 支持Batch on Streaming和Streaming
• Fault-tolerant：高容错，可靠性，checkpoint
• Scalable：可扩展性，1000个节点以上；
• Performance，高性能，高吞吐，低延迟；
• 支持带有事件时间对窗口操作；支持灵活对窗口操作：time、count、session、以及data-driven窗口操作；
• 支持具有Backpressure功能对持续流模型；
• 支持基于轻量级分布式快照(Snapshot实现容错)
• 支持迭代计算
• 支持系统自动化 ：避免特定情况下Shuffle、排序等昂贵操作，中间结果有必要进行缓存
• 低延迟：ms级别时延处理能力；
• Exactly Once：支持有状态计算等Exactly-once语义；提供流异步快照机制，保证流所有数据真正处理一次；
• HA：JobManager支持主备模式，保证无单点故障；
• 水平扩展能力：TaskManager支持手动水平扩展；

Flink四大核心

• Checkpoint ：快照；高容错
• state ：Exactly Once有状态计算
• Time ：通过WaterMark支持基于Event Time的时间窗口
• Window ：窗口机制

数据流：有界流和无界流

Flink认为所有类型等数据都是作为事件流产生等，所有这些数据可以分为有界流和无界流：

• 有界流：

  • 具有起始时间和截止时间，它可以在执行任何等计算之前，先通过摄取所有数据后再来处理有界流；处理有界流不需要有序摄取，因为可以对有界数据集进行排序，有界流对处理既是批处理(Batch)

• 无界流：

  • 它有开始时间但是没有截止时间，它们在生成时提供数据，但不会被终止，无界流必须连续处理数据，即必须在摄取事件后立即处理处理事件，它无法等待所有输入数据到达，因为输入是无界的，在任何时间点都不会完成，处理无界数据通常要求以特定顺序摄取事件，例如事件发生的顺序，以便能够推断结果完整性；

图片1

Window

在流处理中，数据是连续不断的，因此我们不可能等到所有数据都到了才开始处理，当然我们可以每来一个消息就处理一次，但是有时我们需要做一些聚合类的处理，例如，在过去的一分钟内有多少用户点击类我们的网页；这种情况下，我们必须定义一个用来收集最近最近一分钟内的数据,并对这个窗口内的数据进行计算；

窗口可以是时间驱动(Time Window例如30秒)，也可以是数据驱动的(Count Window),一种典型的窗口分类可以分成：防滚窗口(Tumbing Window,无重叠)，滚动窗口(Sliding Window，有重叠)，和会话窗口(Session Window,活动窗口)；

---

• Time Window
• Count Window

---

• Tumbing Window
• Sliding Window
• Session Window

Window

上图中raw data stream代表用户的购买行为流，圈中的数字代表该用户本次购买的商品个数，事件是按时间分布的，所以可以看出事件之间是有time gap的。Flink提供流上图中所有的窗口类型；

Time Window

Flink提出流三种时间概念，分别是event(事件时间：事件发生的时间)、ingestion time(摄取时间：时间进入流处理系统的时间)、processing time(处理时间：消息被计算处理的时间)；Flink中窗口机制和时间类型是完全解耦的，也就是说当需要改变时间类型时不需要更改窗口逻辑相关的代码；

• Tumbling Time Window
  如上图所示，我们需要统计每一分钟中用户购买的商品的总数，需要将用户的行为事件按每一分钟进行切分，成为滚动时间窗口(Tumbling Time Window);翻滚窗口能将数据流切分成不重叠的窗口，每一个事件只能属于一个窗口；

//Stream of(userId, buyCnt)
val buyCnts : DataStream[(Int, Int)] = ...

val tumblingCnts : DataStream[(Int, Int)] = buyCnts
//key stream by userId
.keyBy(0)
//tumbling time window of 1 minute length
// compute sum over buyCnt
.sum(1)

• Sliding Time Window
  对于某些应用，它们需要的窗口是不间断的，需要平滑地进行窗口聚合；比如，我们可以每30秒计算一次最近一分钟用户购买的商品总数；这种窗口我们称为滑动时间窗口(Sliding Time Window)；在滑动窗口中，一个元素可以对应多个窗口；

val slidingCnts : DataStream[Int, Int] = buyCnts.keyBuy(0)
// sliding time window of 1 minute length and 30 secs trigger interval
.timeWindow(Time.minutes(1), Time.seconds(30))
.sum(1)

Count Window

Count Window 是根据元素个数对数据流进行分组的

• Tumbling Count Window
  当我们想要100个用户购买行为事件统计购买总数，那么每当窗口中填满100个元素了，就会对窗口进行计算，这种窗口我们称之为滚动计数窗口：

// Stream of (userId, buyCnts)
val buyCnts: DataStream[(Int, Int)] = ...

val tumblingCnts: DataStream[(Int, Int)] = buyCnts
//Key stream by sensorId
.keyBy(0)
// tumbling count window of 100 elements size
.countWindow(100)
// compute the buyCnt sum
.sum(1)

• Sliding Count Window

CountWindow也支持Silding Window，和Sliding Time window类似：

val slidingCnts: DataStream[(Int, Int)] = vehicleCnts
.keyBy(0)
// sliding count window of 100 elements size and 10 elements trigger interval
.countWindow(100, 10)
.sum(1)

Session Window

在这种用户交互事件流中，我们首先想到的是将事件聚合到会话窗口中(一段用户持续活跃的周期)，由非活跃的间隙分隔开，如上图所示，就是需要计算每个用户在活跃期间总共购买的商品数量，如果用户30秒没有活动则视为会话断开(假设让raw data stream是单个用户的购买行为流)：

// Stream of (userIdm, buyCnts)
val buyCnts: DataStream[(Int, Int)] = ...

val sessionCnts: DataStream[(Int, Int)] = vehicleCnts
.keyBy(0)
// session window based on a 30 seconds session gap interval
.window(ProcessingTimeSessionWindows.withGap(Time.seconds(30)))
.sum(1)

一般而言，window是在无限流上定义了一个有限元素集合，这个集合可以是基于时间的，元素的个数的，时间和个数结合的，会话间隙的，或者是自定义的；F林肯的DataStreamAPI提供了简洁的算子来满足常用的窗口操作，同时提供来通用的窗口机制来允许用户自己定义窗口分配逻辑；

Window API

得益于Flink Window API松耦合设计，我们可以非常灵活地定义符合特定业务的窗口，Flink中定义一个窗口主要需要三个组件：

• Window Assigner：用来决定某个元素被分配到那个/那些窗口中去；下图为目前内置实现的Window Assigners

WindowAssigner

• Trigger：触发器；决定一个窗口何时能够被计算或清除，每个窗口都会拥有一个自己的Trigger；

Trigger

• Evictor: "驱逐者"；在Trigger触发之后，在窗口被处理之前，Evictor(如果有Evictor)会用来剔除窗口中不需要的元素；相当于一个filter

  
  Evictor

上述的三个组件的不同实现组合，可以定义出非常复杂的窗口，Flink中内置的窗口也都是基于这三个组件构成的；

State(只有每一个单独的事件进行转换操作的应用才不需要state)

每一个具有一定复杂度对流处理应用都是又状态对，任何运行基本业务逻辑对流处理应用都需要在一定时间内存储所接收的事件或中间结果，以供后续的某个时间点进行访问并进行后续处理：

• 多种状态基础类型：例如Value、list、map
• Pluggable(插件化) State Backends: State Basckend负责管理应用程序状态，并且在需要的时候进行chechpoint；Flink支持多种state backend，可以将state存在内存或者RocksDB；
• Exactly-once：Flink的checkpoint和故障恢复算法保证来故障发生后应用状态的一致性，因此Flink能够在Application发生故障时对应用程序透明，不造成正确性对影响；
• 超大数据量State：Flink能够利用其异步以及增量式的checkpoint算法，存储TB级别对应用状态；
• Scalable Applications：Flink能够通过在工作节点上对State进行重新分布，支持有状态Application对分布式横向伸缩；

Exactly-once语义保证

Exactly-once语义是Flink的特性之一：是否以为这每一份到达FLink的数据，只会被处理一次；官网的描述如下

Exactly-once state consistency: Flink’s checkpointing and recovery algorithms guarantee the consistency of application state in case of a failure. Hence, failures are transparently handled and do not affect the correctness of an application.

Flink 的 checkpoint 和故障恢复算法保证了故障发生后应用状态的一致性。因此，Flink 能够在应用程序发生故障时，对应用程序透明，不造成正确性的影响。

可以看出，Exactly-once是为有状态的计算准备的

没有状态的算子系统，Flink无法也无需保障只被处理Exactly-once！在失败的情况下，无状态的opertor(map,filter等)只需要数据重新计算一遍即可，例如：

dataStream.filter(_.isInNYC)

当机器(节点)等失败时，只需从最近的一份快照开始，利用可重发的数据源重发一次数据即可，当数据经过filter算子时，全部重新算一次即可，根本不需要区分哪个数据被计算过，哪个数据没有被计算过，因为没有状态的算子只有输入和输出，没有状态可以保存；

此外Flink的Exactly-once需要从最近的一份快照开始重放数据，因此这也和数据源的能力有关，不是所有的数据源都可以提供Exactly-once语义的，以下是官网列出的数据源和Exactly-once语义保障能力列表

Source	Guarantees	Notes
Apache Kafka	exactly once	Use the appropriate Kafka connector for your version
RabbitMQ	at most once(v0.10)/exactly once(v1.0)
Twitter Streaming API	at most once
Collections	exactly once
Files	exactly once
Sockets	exactly once

Flink对于mini-batch（Spark Streaming）的Exactly-once语义的优势

mini-batch的处理过程

• 在数据流中收集记录；
• 收集若干记录后，调度一个批处理作业数据处理；
• 在批处理运行的同时，收集下一批次的记录；

也就是说Spark为了处理一个mini-batch，需要调度一个批处理动作，相比Flink延迟较大，spark的处理是秒级；

而Flink只需要启动一个流计算拓扑，处理持续不断的数据，Flink的处理延迟在毫秒级别，如果涉及到计算中的有多个网络shuffle，SparkStreaming和Flink之间的延迟插件会进一步拉大；

2.Exactly-once语义实现原理

Flink实现Exactly-once语义的原理与SparkStreaming是不一样的；

流处理在计算中既要保障高性能的同时又要保证容错是非常困难的，在批处理中，当作业失败时，可以容易地重新运行作业的失败部分来重新计算丢失的结果，这在批处理中是可行的，因为文件可以从头到尾重放，但是在流处理中却不能这样处理，数据流是无穷无尽的，没有开始和结束点，带有缓冲的数据流可以进行重放一小段数据，但是重最开始重放数据流是不切实际的(流处理作业可能以及运行流数月流)；如果当前的流计算是有状态的，那就意味着除了输出之外没，系统还可以备份和恢复中间算子状态；

记录确认机制(Apache Storm)：

• 开源框架中第一个广泛使用的大规模流处理框架可能是Apache Storm；Storm使用上游备份的记录确认机制来保证失败后重新处理消息，Storm本身不保证状态一致性，任何可变状态的处理都需要委托给用户处理(Storm的Trident API可以确保状态一致性)，storm的无状态模块设计和纯记录确认体系，不适合进行有状态流计算，也无法提供Exactly-once

mini-batch(Storm Trident,Spark Streaming)：

容错流行架构的下一个发展阶段是微批处理或离散化流：为了解决连续计算模型所带来的级别同步的复杂性和开销，连续计算分解为一系列的原子性的批处理，每个batch都可能成功或失败，如果发生故障，重新计算最近的batch；基于微批处理可以实现Exactly-once、高吞吐，但是也有不足：

- 由于是batch，用户不能再任意时间而只能在checkpoint间隔的倍数上的窗口化数据；并且模型不支持许多应用程序所需的计数或会话窗口；

- 微批处理在下游操作比在划分批次的算子(通常指源)中花费更长的时间，则微批次将花费比配置更长的时间没，这导致越来越多的批次排队，或者导致微批量增加

- 微批处理相对于流式处理有更高的延迟，特别是在具有多个网络Shuffle的程序中，很容易将延迟时间延长到数秒；

事务更新(Google Cloud Dataflow)：

在保留连续算子模型(低延迟，背压容错，可变状态)的优势同时又保证Exactly-Once的一种强大而优雅的方法，是原子性记录需要处理的数据并更新到状态中，失败后可以从日志中重新恢复状态以及需要处理的记录；

• 在Google Cloud Dataflow中实现了此概念，系统将计算抽闲为一次部署并长期运行的连续算子的DAG：这为低延迟提供了一种自然的流量控制机制，因为中间过程的缓冲可以缓解背压，直到反压到数据源(基于Pull模式的数据源例如：Kafka)；

• 这种架构的容错原理：通过算子的每个中间记录与更新的状态以及后续产生的记录一起创建一个提交记录，该记录以原子性的方法追加到事务日志或插入到数据库中，在失败的情况下，重放部分数据库日志来恢复计算状态、以及重放丢失的记录；

分布式快照(Flink)

分布式快照与事务更新相比，是将拓扑的状态作为一个整体进行快照，从而减少对分布式存储的写入和频率(确定当前流式计算所处的状态即正在处理中记录和算子状态)，然后生成该状态的一致性快照，并将快照存储在持久化存储中；

• 类似于微批处理，两个检查点之间的所有计算都作为一个原子整体，要么全部成功，要么全部失败，但是我们不必在流处理中暂停来调度下一个微批处理，相反常规数据处理一直运行，数据到达就会处理，而检查点发生在后台；
• Flink将应用程序与流量控制和吞吐量控制分开，更改快照间隔对流处理对结果完全没有影响；

Time

因为事件总是在特定对时间点发生，所以大多数对事件流拥有事件本身所固有对时间语义；进一步而言，许多常见对流计算都基于时间语义，例如：窗口聚合、会话计算、模式检测和基于时间的join；事件时间(event-time)和处理时间(procession-time)

• event-time模式: 使用事件时间语义的流处理应用根据事件本身自带的时间戳进行结果的计算，因此，无论处理的是历史记录的事件还是实时的事件，事件时间模式的处理总能保证结果的准确性和一致性；
• WaterMark：用以衡量事件时间进展，WaterMark也是一种平衡处理延时和完整性的灵活机制；
• Late Data处理: 当以watermark的事件时间模式处理数据流时，在计算完成之后仍会有相关数据到达，这样的事件被称为迟到事件(Late Data)；FLink提供流多种处理迟到数据的选项：例如将这些数据重定向到旁路输出(side output)或更新之前完成计算的结果;
• 处理时间模式(porcession-time)：处理时间模式根据处理引擎的机器时钟触发计算，一般适用于有着严格的低延迟需求，并且能够容忍近似结果的流处理应用；