Apache Flink的内存管理

JVM：

• JAVA本身提供了垃圾回收机制来实现内存管理
• 现今的GC（如Java和.NET）使用分代收集（generation collection），依照对象存活时间的长短使用不同的垃圾收集算法，以达到最好的收集性能。 以Java为例，整个Java堆可以切割成为三个部分：
• Young：
  • Eden：存放新生对象。
  • Survivor：存放经过垃圾回收没有被清除的对象。
  • semi-Spaces：和Survivor做Copying collection。
• Tenured：对象多次回收没有被清除，则移到该区块。
• Perm：存放加载的类别还有方法对象。

Java不同的世代使用不同的GC算法。

• Minor collection：
  • YOUNG世代使用将Eden还有Survivor内的数据利用semi-space做复制收集（Copying collection），
  • 并将原本Survivor内经过多次垃圾收集仍然存活的对象移动到Tenured。
• Major collection则会进行Minor collection，Tenured世代则进行标记压缩收集。

JVM存在的问题：

• Java 对象存储密度低。一个只包含 boolean 属性的对象占用了16个字节内存：对象头占了8个，boolean 属性占了1个，对齐填充占了7个。而实际上只需要一个bit。
• 在处理大量数据时会生成大量对象，Java GC可能会被反复触发，其中Full GC或Major GC的开销是非常大的，GC 会达到秒级甚至分钟级。
• OOM 问题影响稳定性。OutOfMemoryError是分布式计算框架经常会遇到的问题，当JVM中所有对象大小超过分配给JVM的内存大小时，就会发生OutOfMemoryError错误，导致JVM崩溃，分布式框架的健壮性和性能都会受到影响。

Flink的内存管理：

• Flink 并不是将大量对象存在堆上，而是将对象都序列化到一个预分配的内存块上，这个内存块叫做 MemorySegment，它代表了一段固定长度的内存（默认大小为 32KB），也是 Flink 中最小的内存分配单元，并且提供了非常高效的读写方法。每条记录都会以序列化的形式存储在一个或多个MemorySegment中。
• Flink堆内存划分：

• Network Buffers: 一定数量的32KB大小的缓存，主要用于数据的网络传输。在 TaskManager 启动的时候就会分配。默认数量是 2048 个，可以通过 taskmanager.network.numberOfBuffers 来配置
• Memory Manager Pool: 这是一个由 MemoryManager 管理的，由众多MemorySegment组成的超大集合。Flink 中的算法（如 sort/shuffle/join）会向这个内存池申请 MemorySegment，将序列化后的数据存于其中，使用完后释放回内存池。默认情况下，池子占了堆内存的 70% 的大小。
• Remaining (Free) Heap: 这部分的内存是留给用户代码以及 TaskManager 的数据结构使用的，可以把这里看成的新生代。

• 序列化与反序列化可以理解为编码与解码的过程。序列化以后的数据希望占用比较小的空间，而且数据能够被正确地反序列化出来。为了能正确反序列化，序列化时仅存储二进制数据本身肯定不够，需要增加一些辅助的描述信息。此处可以采用不同的策略，因而产生了很多不同的序列化方法。Java本身自带的序列化和反序列化的功能，但是辅助信息占用空间比较大，在序列化对象时记录了过多的类信息。
• Flink实现了自己的序列化框架，Flink处理的数据流通常是一种类型，所以可以只保存一份对象Schema信息，节省存储空间。又因为对象类型固定，所以可以通过偏移量存取。
• Java支持任意Java或Scala类型，类型信息由 TypeInformation 类表示，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: 任意无法匹配之前几种类型的类。
• 针对前六种类型数据集，Flink皆可以自动生成对应的TypeSerializer，能非常高效地对数据集进行序列化和反序列化。对于最后一种数据类型，Flink会使用Kryo进行序列化和反序列化。每个TypeInformation中，都包含了serializer，类型会自动通过serializer进行序列化，然后用Java Unsafe接口写入MemorySegments。如下图展示 一个内嵌型的Tuple3<integer,double,person> 对象的序列化过程：

操纵二进制数据：

• Flink 提供了如 group、sort、join 等操作，这些操作都需要访问海量数据。以sort为例。
• 首先，Flink 会从 MemoryManager 中申请一批 MemorySegment，用来存放排序的数据。

• 这些内存会分为两部分，一个区域是用来存放所有对象完整的二进制数据。另一个区域用来存放指向完整二进制数据的指针以及定长的序列化后的key（key+pointer）。将实际的数据和point+key分开存放有两个目的。第一，交换定长块（key+pointer）更高效，不用交换真实的数据也不用移动其他key和pointer。第二，这样做是缓存友好的，因为key都是连续存储在内存中的，可以增加cache命中。排序会先比较 key 大小，这样就可以直接用二进制的 key 比较而不需要反序列化出整个对象。访问排序后的数据，可以沿着排好序的key+pointer顺序访问，通过 pointer 找到对应的真实数据。

Flink使用堆外内存：

• 启动超大内存（上百GB）的JVM需要很长时间，GC停留时间也会很长（分钟级）。使用堆外内存可以极大地减小堆内存（只需要分配Remaining Heap），使得 TaskManager 扩展到上百GB内存不是问题。
• 进行IO操作时，使用堆外内存可以zero-copy，使用堆内内存至少要复制一次。
• 堆外内存在进程间是共享的。