Apache Flink（下简称Flink）项目是大数据解决领域最近冉冉升起的一颗新星，其不同于其余大数据项目的诸多特性吸引了越来越多人的关注。本文将深入分析Flink的少量关键技术与特性，希望能够帮助读者对Flink有更加深入的理解，对其余大数据系统开发者也能有所裨益。本文假设读者已对MapReduce、Spark及Storm等大数据解决框架有所理解，同时熟习流解决与批解决的基本概念。

Flink简介

Flink核心是一个流式的数据流执行引擎，其针对数据流的分布式计算提供了数据分布、数据通信以及容错机制等功能。基于流执行引擎，Flink提供了诸多更高笼统层的API以便客户编写分布式任务：

1. DataSet API， 对静态数据进行批解决操作，将静态数据笼统成分布式的数据集，客户可以方便地使用Flink提供的各种操作符对分布式数据集进行解决，支持Java、Scala和Python。
2. DataStream API，对数据流进行流解决操作，将流式的数据笼统成分布式的数据流，客户可以方便地对分布式数据流进行各种操作，支持Java和Scala。
3. Table API，对结构化数据进行查询操作，将结构化数据笼统成关系表，并通过类SQL的DSL对关系表进行各种查询操作，支持Java和Scala。

此外，Flink还针对特定的应用领域提供了领域库，例如：

• Flink ML，Flink的机器学习库，提供了机器学习Pipelines API并实现了多种机器学习算法。
• Gelly，Flink的图计算库，提供了图计算的相关API及多种图计算算法实现。

Flink的技术栈如图1所示：

Flink技术栈

我额外整理了一份2018年合适程序员学习的大数据资料（Hadoop，spark，kafka，MapReduce，scala，，推荐算法，实时交易监控系统，客户分析行为，推荐系统），需要的关注我并私信：‘大数据’既可取得。

此外，Flink也可以方便地和Hadoop生态圈中其余项目集成，例如Flink可以读取存储在HDFS或者HBase中的静态数据，以Kafka作为流式的数据源，直接重用MapReduce或者Storm代码，或者是通过YARN申请集群资源等。

统一的批解决与流解决系统

在大数据解决领域，批解决任务与流解决任务一般被认为是两种不同的任务，一个大数据项目一般会被设计为只能解决其中一种任务，例如Apache Storm、Apache Smaza只支持流解决任务，而Aapche MapReduce、Apache Tez、Apache Spark只支持批解决任务。Spark Streaming是Apache Spark之上支持流解决任务的子系统，看似一个特例，实则不然——Spark Streaming采用了一种micro-batch的架构，即把输入的数据流切分成细粒度的batch，并为每一个batch数据提交一个批解决的Spark任务，所以Spark Streaming本质上还是基于Spark批解决系统对流式数据进行解决，和Apache Storm、Apache Smaza等完全流式的数据解决方式完全不同。通过其灵活的执行引擎，Flink能够同时支持批解决任务与流解决任务。

在执行引擎这一层，流解决系统与批解决系统最大不同在于节点间的数据传输方式。对于一个流解决系统，其节点间数据传输的标准模型是：当一条数据被解决完成后，序列化到缓存中，而后立刻通过网络传输到下一个节点，由下一个节点继续解决。而对于一个批解决系统，其节点间数据传输的标准模型是：当一条数据被解决完成后，序列化到缓存中，并不会立刻通过网络传输到下一个节点，当缓存写满，就持久化到本地硬盘上，当所有数据都被解决完成后，才开始将解决后的数据通过网络传输到下一个节点。这两种数据传输模式是两个极端，对应的是流解决系统对低推迟的要求和批解决系统对高吞吐量的要求。Flink的执行引擎采用了一种十分灵活的方式，同时支持了这两种数据传输模型。Flink以固定的缓存块为单位进行网络数据传输，客户可以通过缓存块超时值指定缓存块的传输时机。假如缓存块的超时值为0，则Flink的数据传输方式相似上文所提到流解决系统的标准模型，此时系统可以取得最低的解决推迟。假如缓存块的超时值为无限大，则Flink的数据传输方式相似上文所提到批解决系统的标准模型，此时系统可以取得最高的吞吐量。同时缓存块的超时值也可以设置为0到无限大之间的任意值。缓存块的超时阈值越小，则Flink流解决执行引擎的数据解决推迟越低，但吞吐量也会降低，反之亦然。通过调整缓存块的超时阈值，客户可根据需求灵活地权衡系统推迟和吞吐量。

Flink执行引擎数据传输模式

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

Flink流解决的容错机制

对于一个分布式系统来说，单个进程或者是节点崩溃导致整个Job失败是经常发生的事情，在异常发生时不会丢失客户数据并能自动恢复才是分布式系统必需支持的特性之一。本节主要详情Flink流解决系统任务级别的容错机制。

批解决系统比较容易实现容错机制，因为文件可以重复访问，当某个任务失败后，重启该任务就可。但是到了流解决系统，因为数据源是无限的数据流，从而导致一个流解决任务执行几个月的情况，将所有数据缓存或者是持久化，留待以后重复访问基本上是不可行的。Flink基于分布式快照与可部分重发的数据源实现了容错。客户可自己设置对整个Job进行快照的时间间隔，当任务失败时，Flink会将整个Job恢复到最近一次快照，并从数据源重发快照之后的数据。Flink的分布式快如实现借鉴了Chandy和Lamport在1985年发表的一篇关于分布式快照的论文，其实现的主要思想如下：

按照客户自己设置的分布式快照间隔时间，Flink会定时在所有数据源中插入一种特殊的快照标记消息，这些快照标记消息和其余消息一样在DAG中流动，但是不会被客户定义的业务逻辑所解决，每一个快照标记消息都将其所在的数据流分成两部分：本次快照数据和下次快照数据。

Flink包含快照标记消息的消息流

快照标记消息沿着DAG流经各个操作符，当操作符解决到快照标记消息时，会对自己的状态进行快照，并存储起来。当一个操作符有多个输入的时候，Flink会将先抵达的快照标记消息及其之后的消息缓存起来，当所有的输入中对应该次快照的快照标记消息一律抵达后，操作符对自己的状态快照并存储，之后解决所有快照标记消息之后的已缓存消息。操作符对自己的状态快照并存储可以是异步与增量的操作，并不需要阻塞消息的解决。分布式快照的流程如图4所示：

Flink分布式快照流程图

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

Exactly-Once是流解决系统需要支持的一个非常重要的特性，它保证每一条消息只被流解决系统解决一次，许多流解决任务的业务逻辑都依赖于Exactly-Once特性。相对于At-Least-Once或者是At-Most-Once, Exactly-Once特性对流解决系统的要求更为严格，实现也更加困难。Flink基于分布式快如实现了Exactly-Once特性。

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

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

Flink流解决的时间窗口

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

对于目前大部分流解决系统来说，时间窗口一般是根据Task所在节点的本地时钟进行切分，这种方式实现起来比较容易，不会产生阻塞。但是可能无法满足某些应用需求，比方：

消息本身带有时间戳，客户希望按照消息本身的时间特性进行分段解决。

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

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

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

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

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标记所有小于该时间戳的消息都已流入，Flink的数据源在确认所有小于某个时间戳的消息都已输出到Flink流解决系统后，会生成一个包含该时间戳的WaterMark，插入到消息流中输出到Flink流解决系统中，Flink操作符按照时间窗口缓存所有流入的消息，当操作符解决到WaterMark时，它对所有小于该WaterMark时间戳的时间窗口数据进行解决并发送到下一个操作符节点，而后也将WaterMark发送到下一个操作符节点。

为了保证能够解决所有属于某个时间窗口的消息，操作符必需等到大于这个时间窗口的WaterMark之后才能开始对该时间窗口的消息进行解决，相对于基于Operator Time的时间窗口，Flink需要占用更多内存，且会直接影响消息解决的推迟时间。对此，一个可能的优化措施是，对于聚合类的操作符，可以提前对部分消息进行聚合操作，当有属于该时间窗口的新消息流入时，基于之前的部分聚合结果继续计算，这样的话，只要缓存中间计算结果就可，无需缓存该时间窗口的所有消息。

对于基于Event Time时间窗口的操作符来说，流入WaterMark的时间戳与当前节点的时钟一致是最简单理想的状况，但是在实际环境中是不可能的，因为消息的乱序以及前面节点解决效率的不同，总是会有某些消息流入时间大于其本身的时间戳，真实WaterMark时间戳与理想情况下WaterMark时间戳的差别称为Time Skew，如图5所示：

WaterMark的Time Skew图

整理了一份2018年合适程序员学习的大数据资料（Hadoop，spark，kafka，MapReduce，scala，，推荐算法，实时交易监控系统，客户分析行为，推荐系统），需要的关注我并私信：‘大数据’既可取得。

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

基于时间戳的排序

在流解决系统中，因为流入的消息是无限的，所以对消息进行排序基本上被认为是不可行的。但是在Flink流解决系统中，基于WaterMark，Flink实现了基于时间戳的全局排序。排序的实现思路如下：排序操作符缓存所有流入的消息，当其接收到WaterMark时，对时间戳小于该WaterMark的消息进行排序，并发送到下一个节点，在此排序操作符中释放所有时间戳小于该WaterMark的消息，继续缓存流入的消息，等待下一个WaterMark触发下一次排序。

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

通过支持基于Event Time的消息解决，Flink扩展了其流解决系统的应用范围，使得更多的流解决任务可以通过Flink来执行。

定制的内存管理

Flink项目基于Java及Scala等JVM语言，JVM本身作为一个各种类型应用的执行平台，其对Java对象的管理也是基于通用的解决策略，其垃圾回收器通过估算Java对象的生命周期对Java对象进行有效率的管理。

针对不同类型的应用，客户可能需要针对该类型应用的特点，配置针对性的JVM参数更有效率的管理Java对象，从而提高性能。这种JVM调优的黑魔法需要客户对应用本身及JVM的各参数有深入理解，极大地提高了分布式计算平台的调优门槛。Flink框架本身理解计算逻辑每个步骤的数据传输，相比于JVM垃圾回收器，其理解更多的Java对象生命周期，从而为更有效率地管理Java对象提供了可能。

JVM存在的问题

Java对象开销

相对于c/c++等更加接近底层的语言，Java对象的存储密度相对偏低，例如[1]，“abcd”这样简单的字符串在UTF-8编码中需要4个字节存储，但采用了UTF-16编码存储字符串的Java则需要8个字节，同时Java对象还有header等其余额外信息，一个4字节字符串对象在Java中需要48字节的空间来存储。对于大部分的大数据应用，内存都是稀缺资源，更有效率地内存存储，意味着CPU数据访问吞吐量更高，以及更少磁盘落地的存在。

对象存储结构引发的cache miss

为了缓解CPU解决速度与内存访问速度的差距[2]，现代CPU数据访问一般都会有多级缓存。当从内存加载数据到缓存时，一般是以cache line为单位加载数据，所以当CPU访问的数据假如是在内存中连续存储的话，访问的效率会非常高。假如CPU要访问的数据不在当前缓存所有的cache line中，则需要从内存中加载对应的数据，这被称为一次cache miss。当cache miss非常高的时候，CPU大部分的时间都在等待数据加载，而不是真正的解决数据。Java对象并不是连续的存储在内存上，同时很多的Java数据结构的数据聚集性也不好。

大数据的垃圾回收

Java的垃圾回收机制一直让Java开发者又爱又恨，一方面它免去了开发者自己回收资源的步骤，提高了开发效率，减少了内存泄漏的可能，另一方面垃圾回收也是Java应用的不定时炸弹，有时秒级甚至是分钟级的垃圾回收极大影响了Java应用的性能和可用性。在时下数据中心，大容量内存得到了广泛的应用，甚至出现了单台机器配置TB内存的情况，同时，大数据分析通常会遍历整个源数据集，对数据进行转换、清洗、解决等步骤。在这个过程中，会产生海量的Java对象，JVM的垃圾回收执行效率对性能有很大影响。通过JVM参数调优提高垃圾回收效率需要客户对应用和分布式计算框架以及JVM的各参数有深入理解，而且有时候这也远远不够。

OOM问题

OutOfMemoryError是分布式计算框架经常会遇到的问题，当JVM中所有对象大小超过分配给JVM的内存大小时，就会出现OutOfMemoryError错误，JVM崩溃，分布式框架的健壮性和性能都会受到影响。通过JVM管理内存，同时试图处理OOM问题的应用，通常都需要检查Java对象的大小，并在某些存储Java对象特别多的数据结构中设置阈值进行控制。但是JVM并没有提供官方检查Java对象大小的工具，第三方的工具类库可能无法精确通用地确定Java对象大小[6]。侵入式的阈值检查也会为分布式计算框架的实现添加很多额外与业务逻辑无关的代码。

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那样数据类型需要继承实现org.apache.hadoop.io.Writable接口），也达到了和Hadoop相似的序列化效率。

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

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

整理了一份2018年合适程序员学习的大数据资料（Hadoop，spark，kafka，MapReduce，scala，，推荐算法，实时交易监控系统，客户分析行为，推荐系统），需要的关注我并私信：‘大数据’既可取得。

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

Flink组合类型序列化

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

整理了一份2018年合适程序员学习的大数据资料（Hadoop，spark，kafka，MapReduce，scala，，推荐算法，实时交易监控系统，客户分析行为，推荐系统），需要的关注我并私信：‘大数据’既可取得。

显式的内存管理

垃圾回收是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管理。

Network buffers在Flink中主要基于Netty的网络传输，无需多讲。Remaining Heap用于UDF中客户自己创立的Java对象，在UDF中，客户通常是流式的解决数据，并不需要很多内存，同时Flink也不鼓励客户在UDF中缓存很多数据，由于这会引起前面提到的诸多问题。Memory Manager pool（以后以内存池代指）通常会配置为最大的一块内存，接下来会详细详情。

在Flink中，内存池由多个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集中。数据序列化后的Key和指向第一个MemorySegment集中值的指针存放于第二个MemorySegment集中。
• 对第二个MemorySegment集中的Key进行排序，如需交换Key位置，只要交换对应的Key+Pointer的位置，第一个MemorySegment集中的数据无需改变。 当比较两个Key大小时，TypeComparator提供了直接基于二进制数据的比照方法，无需反序列化任何数据。
• 排序完成后，访问数据时，按照第二个MemorySegment集中Key的顺序访问，并通过Pointer值找到数据在第一个MemorySegment集中的位置，通过TypeSerializer反序列化成Java对象返回。

Flink排序算法

这样实现的好处有：

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

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

总结

本文主要详情了Flink项目的少量关键特性，Flink是一个拥有诸多特色的项目，包括其统一的批解决和流解决执行引擎，通用大数据计算框架与传统数据库系统的技术结合，以及流解决系统的诸多技术创新等，由于篇幅有限，Flink还有少量其余很有意思的特性没有详细详情，比方DataSet API级别的执行计划优化器，原生的迭代操作符等，感兴趣的读者可以通过Flink官网理解更多Flink的详细内容。希望通过本文的详情能够让读者对Flink有更多的理解，也让更多的人使用甚至参加到Flink项目中去。

作者简介：李呈祥 ，Intel BigData Team软件工程师，主要关注大数据计算框架与SQL引擎的性能优化，Apache Hive Committer，Apache Flink Contributor。