统一的批处理与流处理系统

在大数据处理领域，批处理任务与流处理任务一般被认为是两种不同的任务，一个大数据项目一般会被设计为只能处理其中一种任务，

• Apache Storm、Apache Smaza只支持流处理任务，
• Aapche MapReduce、Apache Tez、Apache Spark只支持批处理任务。

通过其灵活的执行引擎，Flink能够同时支持批处理任务与流处理任务。

在执行引擎这一层，流处理系统与批处理系统最大不同：节点间的数据传输方式：

• 流处理系统：其节点间数据传输的标准模型是，当一条数据被处理完成后，序列化到缓存中，
  然后立刻网络传输到下一个节点，继续处理。
• 批处理系统，其节点间数据传输的标准模型是，当一条数据被处理完成后，序列化到缓存中，并不会立刻通过网络传输到下一个节点，
  当缓存写满，就持久化到本地硬盘上，
  当所有数据都被处理完成后，才开始将处理后的数据通过网络传输到下一个节点。

这两种数据传输模式是两个极端，对应的是流处理系统对低延迟的要求和批处理系统对高吞吐量的要求。

Flink的执行引擎同时支持了这两种数据传输模型。

Flink以固定的缓存块为单位进行网络数据传输，用户可以通过缓存块超时值，指定缓存块的传输时机：

• 如果缓存块的超时值为0，则Flink的数据传输方式是流处理系统的标准模型，系统可以获得最低的处理延迟。
• 如果缓存块的超时值为无限大，则Flink的数据传输方式是批处理系统的标准模型，此时系统可以获得最高的吞吐量。
• 缓存块的超时值也可以设置为0到无限大之间的任意值。
  缓存块的超时阈值越小，则Flink流处理执行引擎的数据处理延迟越低，但吞吐量也会降低，反之亦然。

通过调整缓存块的超时阈值，用户可根据需求灵活地权衡系统延迟和吞吐量。

在统一的流式执行引擎基础上，Flink同时支持了流计算和批处理，并对性能（延迟、吞吐量等）有所保障。
相对于其他原生的流处理与批处理系统，并没有因为统一执行引擎而受到影响，从而大幅度减轻了用户安装、部署、监控、维护等成本。

Flink流处理的容错机制

批处理系统比较容易实现容错机制，由于文件可以重复访问，当某个任务失败后，重启该任务即可。
Flink基于分布式快照和可部分重发的数据源实现了容错。
用户可自定义对整个Job进行快照的时间间隔。
当任务失败时，Flink会将整个Job恢复到最近一次快照，并从数据源重发快照之后的数据。
Flink的分布式快照实现借鉴了Chandy和Lamport在1985年发表的一篇关于分布式快照的论文，其实现的主要思想如下：
按照用户自定义的分布式快照间隔时间，Flink会定时在所有【数据源中】插入一种特殊的快照标记消息，
这些快照标记消息和其他消息一样在DAG中流动，但是不会被用户定义的业务逻辑所处理，
每一个快照标记消息都将其所在的数据流分成两部分：本次快照数据和下次快照数据。

快照标记消息沿着DAG流经各个操作符，当操作符处理到快照标记消息时，会对自己的状态进行快照，并存储起来。
当一个操作符有多个输入时，Flink会将先抵达的快照标记消息，及其之后的消息缓存起来，
当所有的输入中【对应】该次快照的快照标记消息全部抵达后，操作符对自己的状态快照并存储，之后处理所有快照标记消息之后的已缓存消息。
操作符对自己的状态快照并存储可以是异步与增量的操作，并不需要阻塞消息的处理。

当所有的Data Sink（终点操作符）都收到快照标记信息并对自己的状态快照和存储后，整个分布式快照就完成了，同时通知数据源，释放该快照标记消息之前的所有消息。
若之后发生节点崩溃等异常情况时，只需要恢复之前存储的分布式快照状态，并从数据源重发该快照以后的消息就可以了。

Flink基于分布式快照实现了Exactly-Once特性。

相对于其他流处理系统的容错方案，Flink基于分布式快照的方案在功能和性能方面都具有很多优点，包括：

• 低延迟。由于操作符状态的存储可以异步，所以进行快照的过程基本上不会阻塞消息的处理，因此不会对消息延迟产生负面影响。
• 高吞吐量。当操作符状态较少时，对吞吐量基本没有影响。当操作符状态较多时，相对于其他的容错机制，分布式快照的时间间隔是用户自定义的，所以用户可以权衡错误恢复时间和吞吐量要求来调整分布式快照的时间间隔。
• 与业务逻辑的隔离。Flink的分布式快照机制与用户的业务逻辑是完全隔离的，用户的业务逻辑不会依赖或是对分布式快照产生任何影响。
• 错误恢复代价。分布式快照的时间间隔越短，错误恢复的时间越少，与吞吐量负相关。

Flink流处理的时间窗口

对于流处理系统来说，流入的消息不存在上限，所以对于聚合或是连接等操作，流处理系统需要对流入的消息进行分段，然后基于每一段数据进行聚合或是连接。
消息的分段即称为窗口，流处理系统支持的窗口有很多类型，最常见的就是时间窗口，基于时间间隔对消息进行分段处理。

由于不同节点的时钟可能不同，以及消息在流经各个节点的延迟不同，在某个节点属于同一个时间窗口处理的消息，流到下一个节点时可能被切分到不同的时间窗口中，从而产生不符合预期的结果。

Flink支持3种类型的时间窗口，分别适用于用户对于时间窗口不同类型的要求：

1. Operator Time。根据Task所在节点的本地时钟来切分的时间窗口。

2. Event Time。消息自带时间戳，根据消息的时间戳进行处理，确保时间戳在同一个时间窗口的所有消息一定会被正确处理。由于消息可能乱序流入Task，所以Task需要缓存当前时间窗口消息处理的状态，直到确认属于该时间窗口的所有消息都被处理，才可以释放，如果乱序的消息延迟很高会影响分布式系统的吞吐量和延迟。

3. Ingress Time。有时消息本身并不带有时间戳信息，但用户依然希望按照消息而不是节点时钟划分时间窗口，例如避免上面提到的第二个问题，此时可以在消息源流入Flink流处理系统时自动生成增量的时间戳赋予消息，之后处理的流程与Event Time相同。
   Ingress Time可以看成是Event Time的一个特例，由于其在消息源处时间戳一定是有序的，所以在流处理系统中，相对于Event Time，其乱序的消息延迟不会很高，因此对Flink分布式系统的吞吐量和延迟的影响也会更小。

Event Time时间窗口的实现

Flink借鉴了Google的MillWheel项目，通过WaterMark来支持基于Event Time的时间窗口。

当操作符通过基于Event Time的时间窗口来处理数据时，它必须在确定所有属于该时间窗口的消息全部流入此操作符后才能开始数据处理。
但是由于消息可能是乱序的，所以操作符无法直接确认何时所有属于该时间窗口的消息全部流入此操作符。
WaterMark包含一个时间戳，Flink使用WaterMark标记所有小于该时间戳的消息都已流入：(和上面的快照的实现类似)

1. Flink的数据源，在确认所有小于某个时间戳的消息，都已输出到Flink流处理系统后，会生成一个包含该时间戳的WaterMark，插入到消息流中，输出到Flink流处理系统中，
2. Flink操作符按照时间窗口缓存所有流入的消息，
3. 当操作符处理到WaterMark时，它对所有小于该WaterMark时间戳的时间窗口数据进行处理，并发送到下一个操作符节点，
4. 然后也将WaterMark发送到下一个操作符节点。

为了保证能够处理所有属于某个时间窗口的消息，操作符必须等到【大于这个时间窗口的WaterMark之后】才能开始对该时间窗口的消息进行处理，

• 相对于基于Operator Time的时间窗口，Flink需要占用更多内存，且会直接影响消息处理的延迟时间。
  对此一个可能的优化措施是：对于聚合类操作符，可提前对部分消息聚合，
  当有该时间窗口的新消息流入时，基于之前的部分聚合结果计算，这样只需缓存中间计算结果，无需缓存该时间窗口的所有消息。
• 对于基于Event Time时间窗口的操作符来说，流入WaterMark的时间戳与当前节点的时钟一致，在实际环境中是不可能的，
  由于消息的乱序以及前面节点处理效率的不同，总是会有某些消息流入时间大于其本身的时间戳，
  真实WaterMark时间戳与理想情况下WaterMark时间戳的差别称为Time Skew

Time Skew决定了该WaterMark与上一个WaterMark之间的时间窗口所有数据需要缓存的时间，Time Skew时间越长，该时间窗口数据的延迟越长，占用内存的时间也越长，同时会对流处理系统的吞吐量产生负面影响。

基于时间戳的排序（基于watermark实现）

在流处理系统中，由于流入的消息是无限的，所以对消息进行排序基本上被认为是不可行的。
但是在Flink流处理系统中，基于WaterMark，Flink实现了基于时间戳的全局排序。排序的实现思路如下：

1. 排序操作缓存所有流入的消息，当其接收到WaterMark时，对时间戳小于该WaterMark的消息进行排序，
2. 发送到下一个节点，
3. 在此排序操作符中，释放所有时间戳小于该WaterMark的消息，继续缓存流入的消息，等待下一个WaterMark触发下一次排序。

由于WaterMark保证了在其之后不会出现时间戳比它小的消息，所以可以保证排序的正确性。
需要注意的是，如果排序操作符有多个节点，只能保证每个节点的流出消息是有序的，节点之间的消息不能保证有序，要实现全局有序，则只能有一个排序操作符节点。

定制的内存管理

JVM存在的问题

• Java对象开销
• 对象存储结构引发的cache miss
• 大数据的垃圾回收
• OOM问题

Flink的处理策略

为了解决以上提到的问题，高性能分布式计算框架通常需要以下技术：

• 定制的序列化工具。
  显式内存管理的前提步骤就是序列化，将Java对象序列化成二进制，数据存储在内存上（on heap或是off-heap）。
  通用的序列化框架，如Java默认使用java.io.Serializable将Java对象及其成员变量的所有元信息，作为其序列化数据的一部分，序列化后的数据包含了所有反序列化所需的信息。
  对于Flink这样的分布式计算框架来说，这些元数据信息可能是冗余数据。
  定制的序列化框架，如Hadoop的org.apache.hadoop.io.Writable需要用户实现该接口，并自定义类的序列化和反序列化方法。
  这种方式效率最高，但需要用户额外的工作，不够友好。
  直接通过偏移量，从而只需要反序列化特定的对象成员变量。

• 显式的内存管理。
  一般通用的做法是批量申请和释放内存，每个JVM实例有一个统一的内存管理器，所有内存的申请和释放都通过该内存管理器进行。
  这可以避免常见的内存碎片问题，同时由于数据以二进制的方式存储，可以大大减轻垃圾回收压力。

• 缓存友好的数据结构和算法。
  • 对于计算密集的数据结构和算法，直接操作序列化后的二进制数据，而不是将对象反序列化后再进行操作。
  • 同时，只将操作相关的数据连续存储，与操作无关的数据放弃，可以最大化的利用L1/L2/L3缓存，减少Cache miss的概率，提升CPU计算的吞吐量。
    以排序为例，由于排序的主要操作是对Key进行对比，如果将所有排序数据的Key与Value分开并对Key连续存储，那么访问Key时的Cache命中率会大大提高。

定制的序列化工具

分布式计算框架可以使用定制序列化工具的前提，是要待处理数据流通常是同一类型，由于数据集对象的类型固定，从而可以只保存一份对象Schema信息，节省大量的存储空间。
同时，对于固定大小的类型，也可通过固定的偏移位置存取。
在需要访问某个对象成员变量时，通过定制的序列化工具，并不需要反序列化整个Java对象，而是直接通过偏移量，从而只需要反序列化特定的对象成员变量。
如果对象的成员变量较多时，能够大大减少Java对象的创建开销，以及内存数据的拷贝大小。
Flink数据集都支持任意Java或是Scala类型，通过自动生成定制序列化工具，既保证了API接口对用户友好，也达到了和Hadoop类似的序列化效率。

Flink对数据集的类型信息进行分析，然后自动生成定制的序列化工具类。
Flink支持任意的Java或是Scala类型，
通过Java Reflection框架分析基于Java的Flink程序UDF（User Define Function）的返回类型的类型信息，
通过Scala Compiler分析基于Scala的Flink程序UDF的返回类型的类型信息。
类型信息由TypeInformation类表示，这个类有诸多具体实现类，例如：

1. BasicTypeInfo：任意Java基本类型（装包或未装包）和String类型。
2. BasicArrayTypeInfo：任意Java基本类型数组（装包或未装包）和String数组。
3. WritableTypeInfo：任意Hadoop的Writable接口的实现类。
4. TupleTypeInfo：任意的Flink tuple类型(支持Tuple1 to Tuple25)。 Flink tuples是固定长度固定类型的Java Tuple实现。
5. CaseClassTypeInfo：任意的 Scala CaseClass(包括 Scala tuples)。
6. PojoTypeInfo：任意的POJO (Java or Scala)，例如Java对象的所有成员变量，要么是public修饰符定义，要么有getter/setter方法。
7. GenericTypeInfo：任意无法匹配之前几种类型的类。

前6种类型数据集几乎覆盖了绝大部分的Flink程序，针对前6种类型数据集，Flink皆可以自动生成对应的TypeSerializer定制序列化工具，非常有效率地对数据集进行序列化和反序列化。对于第7种类型，Flink使用Kryo进行序列化和反序列化。
此外，对于可被用作Key的类型，Flink还同时自动生成TypeComparator，用来辅助直接对序列化后的二进制数据直接进行compare、hash等操作。
对于Tuple、CaseClass、Pojo等组合类型，Flink自动生成的TypeSerializer、TypeComparator同样是组合的，并把其成员的序列化/反序列化代理给其成员对应的TypeSerializer、TypeComparator，如图6所示：

此外如有需要，用户可通过集成TypeInformation接口定制实现自己的序列化工具。

显式的内存管理

垃圾回收是JVM内存管理回避不了的问题，JDK8的G1算法改善了JVM垃圾回收的效率和可用范围，但对于大数据处理实际环境还远远不够。
这也和现在分布式框架的发展趋势有所冲突，越来越多的分布式计算框架希望尽可能多地将待处理数据集放入内存，
而对于JVM垃圾回收来说，内存中Java对象越少、存活时间越短，其效率越高。
通过JVM进行内存管理的话，OutOfMemoryError也是一个很难解决的问题。
同时，在JVM内存管理中，Java对象有潜在的碎片化存储问题（Java对象所有信息可能在内存中连续存储），也有可能在所有Java对象大小没有超过JVM分配内存时，出现OutOfMemoryError问题。
Flink将内存分为3个部分，每个部分都有不同用途：

• Network buffers： 一些以32KB Byte数组为单位的buffer，主要被网络模块用于数据的网络传输。
• Memory Manager pool：大量以32KB Byte数组为单位的内存池，
  所有的运行时算法（例如Sort/Shuffle/Join）都从这个内存池申请内存，
  并将序列化后的数据存储其中，结束后释放回内存池。
• Remaining (Free) Heap：主要留给UDF中用户自己创建的Java对象，由JVM管理。
  在UDF中，用户通常是流式的处理数据，并不需要很多内存，同时Flink也不鼓励用户在UDF中缓存很多数据。

内存池Memory Manager pool由多个MemorySegment组成，
每个MemorySegment代表一块连续的内存，底层存储是byte[]，默认32KB大小。
MemorySegment提供了根据偏移量访问数据的各种方法，如get/put int、long、float、double等，
MemorySegment之间数据拷贝等方法和java.nio.ByteBuffer类似。
对于Flink的数据结构，通常包括多个向内存池申请的MemeorySegment，所有要存入的对象通过TypeSerializer序列化之后，将二进制数据存储在MemorySegment中，
在取出时通过TypeSerializer反序列化。
数据结构通过MemorySegment提供的set/get方法访问具体的二进制数据。Flink这种看起来比较复杂的内存管理方式带来的好处主要有：

• 二进制的数据存储大大提高了数据存储密度，节省了存储空间。
• 所有的运行时数据结构和算法只能通过内存池申请内存，保证了其使用的内存大小是固定的，不会因为运行时数据结构和算法而发生OOM。对于大部分的分布式计算框架来说，这部分由于要缓存大量数据最有可能导致OOM。
• 内存池虽然占据了大部分内存，但其中的MemorySegment容量较大（默认32KB），所以内存池中的Java对象其实很少，而且一直被内存池引用，所有在垃圾回收时很快进入持久代，大大减轻了JVM垃圾回收的压力。
• Remaining Heap的内存虽然由JVM管理，但是由于其主要用来存储用户处理的流式数据，生命周期非常短，速度很快的Minor GC就会全部回收掉，一般不会触发Full GC。

Flink当前的内存管理在最底层是基于byte[]，所以数据最终还是on-heap，最近Flink增加了off-heap的内存管理支持。Flink off-heap的内存管理相对于on-heap的优点主要在于：

• 启动分配了大内存(例如100G)的JVM很耗费时间，垃圾回收也很慢。如果采用off-heap，剩下的Network buffer和Remaining heap都会很小，垃圾回收也不用考虑MemorySegment中的Java对象了。
• 更有效率的IO操作。在off-heap下，将MemorySegment写到磁盘或是网络可以支持zeor-copy技术，而on-heap的话则至少需要一次内存拷贝。
• off-heap可用于错误恢复，比如JVM崩溃，在on-heap时数据也随之丢失，但在off-heap下，off-heap的数据可能还在。此外，off-heap上的数据还可以和其他程序共享。

缓存友好的计算

磁盘IO和网络IO之前一直被认为是Hadoop系统的瓶颈，但是随着Spark、Flink等新一代分布式计算框架的发展，越来越多的趋势使得CPU/Memory逐渐成为瓶颈，这些趋势包括：

• 更先进的IO硬件逐渐普及。10GB网络和SSD硬盘等已经被越来越多的数据中心使用。
• 更高效的存储格式。Parquet，ORC等列式存储被越来越多的Hadoop项目支持，其非常高效的压缩性能大大减少了落地存储的数据量。
• 更高效的执行计划。例如很多SQL系统执行计划优化器的Fliter-Push-Down优化会将过滤条件尽可能的提前，甚至提前到Parquet的数据访问层，使得在很多实际的工作负载中并不需要很多的磁盘IO。

由于CPU处理速度和内存访问速度的差距，提升CPU的处理效率的关键在于最大化的利用L1/L2/L3/Memory，减少任何不必要的Cache miss。
定制的序列化工具给Flink提供了可能，通过定制的序列化工具，Flink访问的二进制数据本身，因为占用内存较小，存储密度比较大，
而且还可以在设计数据结构和算法时尽量连续存储，减少内存碎片化对Cache命中率的影响，
甚至更进一步，Flink可以只是将需要操作的部分数据（如排序时的Key）连续存储，而将其他部分的数据存储在其他地方，从而最大可能地提升Cache命中的概率。

Flink排序算法

Flink通过特殊设计的排序算法获得了非常好的性能，其排序算法的实现如下：

• 将待排序的数据序列化后，用两个不同的MemorySegment集实现：
  • 其中一个MemorySegment集中，存放：数据全部的序列化值。
  • 第二个MemorySegment集中，存放：数据序列化后的Key和指向第一个MemorySegment集中值的指针。
• 对第二个MemorySegment集中的Key进行排序，
  如需交换Key位置，只需交换对应的Key+Pointer的位置，第一个MemorySegment集中的数据无需改变。
  当比较两个Key大小时，TypeComparator提供了直接基于二进制数据的对比方法，无需反序列化任何数据。

• 排序完成后，访问数据时，按照第二个MemorySegment集中Key的顺序访问，
  并通过Pointer值找到数据在第一个MemorySegment集中的位置，通过TypeSerializer反序列化成Java对象返回。

  

这样实现的好处有：

• 通过Key和Full data分离存储的方式，尽量将被操作的数据最小化，提高Cache命中的概率，从而提高CPU的吞吐量。
• 移动数据时，只需移动Key+Pointer，而无须移动数据本身，大大减少了内存拷贝的数据量。
• TypeComparator直接基于二进制数据进行操作，节省了反序列化的时间。

通过定制的内存管理，Flink通过充分利用内存与CPU缓存，大大提高了CPU的执行效率，
同时由于大部分内存都由框架自己控制，也很大程度提升了系统的健壮性，减少了OOM出现的可能。