Flink教程（29）- Flink内存管理-阿里云开发者社区

Flink教程（29）- Flink内存管理

2023-11-20 132

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

本文涉及的产品

实时计算 Flink 版，5000CU*H 3个月

云原生大数据计算服务 MaxCompute，5000CU*H 100GB 3个月

云原生大数据计算服务MaxCompute，500CU*H 100GB 3个月

简介： Flink教程（29）- Flink内存管理

01 引言

在前面的博客，我们学习了Flink的性能优化了，有兴趣的同学可以参阅下：

本文主要讲解Flink的内存管理。

02 Flink内存管理

Flink本身基本是以Java语言完成的，理论上说，直接使用JVM的虚拟机的内存管理就应该更简单方便，但Flink还是单独抽象出了自己的内存管理。

因为Flink是为大数据而产生的，而大数据使用会消耗大量的内存，而JVM的内存管理管理设计是兼顾平衡的，不可能单独为了大数据而修改，这对于Flink来说，非常的不灵活，而且频繁GC会导致长时间的机器暂停应用，这对于大数据的应用场景来说也是无法忍受的。

JVM在大数据环境下存在的问题:

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

2.1 Flink内存划分

注意:Flink的内存管理是在JVM的基础之上,自己进行的管理,但是还没有逃脱的JVM,具体怎么实现,现阶段我们搞不定

网络缓冲区Network Buffers：这个是在TaskManager启动的时候分配的，这是一组用于缓存网络数据的内存，每个块是32K，默认分配2048个，可以通过“taskmanager.network.numberOfBuffers”修改
内存池Memory Manage pool：大量的Memory Segment块，用于运行时的算法（Sort/Join/Shuffle等），这部分启动的时候就会分配。默认情况下，占堆内存的70% 的大小。
用户使用内存Remaining (Free) Heap: 这部分的内存是留给用户代码以及 TaskManager的数据使用的。

2.2 Flink堆外内存

除了JVM之上封装的内存管理,还会有个一个很大的堆外内存,用来执行一些IO操作

启动超大内存（上百GB）的JVM需要很长时间，GC停留时间也会很长（分钟级）。

使用堆外内存可以极大地减小堆内存（只需要分配Remaining Heap），使得 TaskManager 扩展到上百GB内存不是问题。

进行IO操作时，使用堆外内存(可以理解为使用操作系统内存)可以zero-copy，使用堆内JVM内存至少要复制一次(需要在操作系统和JVM直接进行拷贝)。

堆外内存在进程间是共享的。

总结:

Flink相对于Spark,堆外内存该用还是用, 堆内内存管理做了自己的封装,不受JVM的GC影响

2.3 序列化与反序列化

Flink除了对堆内内存做了封装之外,还实现了自己的序列化和反序列化机制，

序列化与反序列化可以理解为编码与解码的过程。序列化以后的数据希望占用比较小的空间，而且数据能够被正确地反序列化出来。为了能正确反序列化，序列化时仅存储二进制数据本身肯定不够，需要增加一些辅助的描述信息。此处可以采用不同的策略，因而产生了很多不同的序列化方法。

Java本身自带的序列化和反序列化的功能，但是辅助信息占用空间比较大，在序列化对象时记录了过多的类信息。

Flink实现了自己的序列化框架，使用TypeInformation表示每种数据类型，所以可以只保存一份对象Schema信息，节省存储空间。又因为对象类型固定，所以可以通过偏移量存取。

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: 任意无法匹配之前几种类型的类。(除了该数据使用kyro序列化.上面的其他的都是用二进制)

针对前六种类型数据集，Flink皆可以自动生成对应的TypeSerializer，能非常高效地对数据集进行序列化和反序列化。对于最后一种数据类型，Flink会使用Kryo进行序列化和反序列化。每个TypeInformation中，都包含了serializer，类型会自动通过serializer进行序列化，然后用Java Unsafe接口(具有像C语言一样的操作内存空间的能力)写入MemorySegments。

Flink通过自己的序列化和反序列化,可以将数据进行高效的存储,不浪费内存空间

2.4 操纵二进制数据

Flink中的group、sort、join 等操作可能需要访问海量数据。以sort为例。

首先，Flink 会从 MemoryManager 中申请一批 MemorySegment，用来存放排序的数据。

这些内存会分为两部分:

一个区域是用来存放所有对象完整的二进制数据。
另一个区域用来存放指向完整二进制数据的指针以及定长的序列化后的key（key+pointer）。

将实际的数据和point+key分开存放有两个目的:

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

在交换过程中，只需要比较key就可以完成sort的过程，只有key1 == key2的情况，才需要反序列化拿出实际的对象做比较，而比较之后只需要交换对应的key而不需要交换实际的对象。

2.5 注意

减少full gc时间：因为所有常用数据都在Memory Manager里，这部分内存的生命周期是伴随TaskManager管理的而不会被GC回收。其他的常用数据对象都是用户定义的数据对象，这部分会快速的被GC回收。
减少OOM：所有的运行时的内存应用都从池化的内存中获取，而且运行时的算法可以在内存不足的时候将数据写到堆外内存
节约空间：由于Flink自定序列化/反序列化的方法，所有的对象都以二进制的形式存储，降低消耗
高效的二进制操作和缓存友好：二进制数据以定义好的格式存储，可以高效地比较与操作。另外，该二进制形式可以把相关的值，以及hash值，键值和指针等相邻地放进内存中。这使得数据结构可以对CPU高速缓存更友好，可以从CPU的 L1/L2/L3 缓存获得性能的提升,也就是Flink的数据存储二进制格式符合CPU缓存的标准,非常方便被CPU的L1/L2/L3各级别缓存利用,比内存还要快!

03 文末

本文主要讲了Flink的内存管理，谢谢大家的阅读，本文完！

Flink教程（29）- Flink内存管理

01 引言