Flink内存管理
1.简介
自从2003-2006年，Google发表了三篇著名的大数据相关论文（Google FS，MapReduce，Big Table）后，内存问题一直困扰大数据工程师们。
这一问题从MR1.0一直延续到Spark时代，从Spark晚期版本试图由应用程序自行管理内存后，人们才初步解决了内存问题。
使用原生的JVM内存管理会带来如下的致命问题：
- JVM对象存储密度低，在32位系统或开启指针压缩的64位系统中，普通对象（非数组）对象头占用64bit，尾部还需要8字节对齐。
- JVM GC导致的毛刺和性能问题，由于计算引擎会频繁创建对象，小对象会被创建在新生代导致频繁的minor GC和STW，大对象会被直接创建在老年代导致大量的并发式GC（CMS）或混合式GC（G1），并且GC的触发和执行完全由JVM控制，计算引擎无法干预。
- 潜在的OOM风险，OOM发生的时机不可控。
在Apache Flink中，taskManager自行管理的内存，避免了JVM原生内存管理的缺陷，本文将详细介绍相关逻辑。
2.内存模型
Task manager管理的JVM内存主要分为Network Buffers、MemoryManager 和 Free 三个区域。
- Network Buffers，shuffle / broadcost网络活动相关的内存
- MemoryManager，cache / sorting / hashing 计算相关的内存
- Free，存放用户代码产生的对象
3.代码分析
3.1 TaskManagerOptions
内存管理的相关配置
- MEMORY_SEGMENT_SIZE——内存段大小，默认32kb。内存段（segment）是Flink内存管理的基本模型。
- MANAGED_MEMORY_SIZE——task manager管理的内存大小，如果不设置则使用MANAGED_MEMORY_FRACTION
- MANAGED_MEMORY_FRACTION——task manager管理的内存占比，默认0.7f
- MEMORY_OFF_HEAP——是否使用堆外内存，默认false即使用堆内内存
- MANAGED_MEMORY_PRE_ALLOCATE——task manager启动时是否预分配，默认false
- NETWORK_NUM_BUFFERS——网络缓冲区的segment数量，默认2048
- NETWORK_BUFFERS_MEMORY_FRACTION——网络缓冲区的内存占比，默认0.1
- NETWORK_BUFFERS_MEMORY_MIN——网络缓冲区的最小size，默认64MB
- NETWORK_BUFFERS_MEMORY_MAX——网络缓冲区的最大size，默认1GB
3.2 MemoryPool
静态抽象类MemoryPool定义了内存池的方法，它有两个实现类HybridHeapMemoryPool和HybridOffHeapMemoryPool，堆内内存池和堆外内存池。
abstract static class MemoryPool {
abstract int getNumberOfAvailableMemorySegments();
abstract MemorySegment allocateNewSegment(Object owner);
abstract MemorySegment requestSegmentFromPool(Object owner);
abstract void returnSegmentToPool(MemorySegment segment);
abstract void clear();
}
3.3 MemoryManager
MemoryManager 类负责管理sorting,、hashing、caching使用的内存，主要方法有allocatePages（申请内存段）和release（释放内存段）
public void allocatePages(Object owner, List<MemorySegment> target, int numPages)
throws MemoryAllocationException {
... 入参校验
// -------------------- BEGIN CRITICAL SECTION -------------------
synchronized (lock) {
if (isShutDown) {
throw new IllegalStateException("Memory manager has been shut down.");
}
// 可用内存校验
if (numPages > (memoryPool.getNumberOfAvailableMemorySegments() + numNonAllocatedPages)) {
throw new MemoryAllocationException("Could not allocate " + numPages + " pages. Only " +
(memoryPool.getNumberOfAvailableMemorySegments() + numNonAllocatedPages)
+ " pages are remaining.");
}
// allocatedSegments是个HashMap<Object, Set<MemorySegment>>，key是owner，值是此owner占用的segment列表
Set<MemorySegment> segmentsForOwner = allocatedSegments.get(owner);
if (segmentsForOwner == null) {
segmentsForOwner = new HashSet<MemorySegment>(numPages);
allocatedSegments.put(owner, segmentsForOwner);
}
if (isPreAllocated) {
for (int i = numPages; i > 0; i--) {
MemorySegment segment = memoryPool.requestSegmentFromPool(owner);
target.add(segment);
segmentsForOwner.add(segment);
}
}
else {
for (int i = numPages; i > 0; i--) {
MemorySegment segment = memoryPool.allocateNewSegment(owner);
target.add(segment);
segmentsForOwner.add(segment);
}
numNonAllocatedPages -= numPages;
}
}
// -------------------- END CRITICAL SECTION -------------------
}
3.4 MemorySegment
MemorySegment类管理Flink中的一个内存页，MemorySegment是抽象类有两个实现类HeapMemorySegment和HybridMemorySegment。
MemorySegment定义了Segment的基本操作：
// 返回segment字节数
public int size();
// segment是否已释放
public boolean isFreed();
// 释放segment
public void free();
// 是否使用堆外内存
public boolean isOffHeap();
// 返回堆内内存的数组
public byte[] getArray();
// 返回堆外内存的地址
public long getAddress();
// 返回指定区域的数据，并封装成ByteBuffer
public abstract ByteBuffer wrap(int offset, int length);
// 返回segment owner
public Object getOwner();
// 随机读写API
public abstract byte get(int index);
public abstract void put(int index, byte b);
public abstract void get(int index, byte[] dst);
public abstract void put(int index, byte[] src);
public abstract void get(int index, byte[] dst, int offset, int length);
public abstract void put(int index, byte[] src, int offset, int length);
public abstract boolean getBoolean(int index);
public abstract void putBoolean(int index, boolean value);
public final char getChar(int index);
public final char getCharLittleEndian(int index);
public final char getCharBigEndian(int index);
public final void putChar(int index, char value);
public final void putCharLittleEndian(int index, char value);
public final void putCharBigEndian(int index, char value);
public final short getShort(int index);
public final short getShortLittleEndian(int index);
public final short getShortBigEndian(int index);
public final void putShort(int index, short value);
public final void putShortLittleEndian(int index, short value);
public final void putShortBigEndian(int index, short value);
public final int getInt(int index);
public final int getIntLittleEndian(int index);
public final int getIntBigEndian(int index);
public final void putInt(int index, int value);
public final void putIntLittleEndian(int index, int value);
public final void putIntBigEndian(int index, int value);
public final long getLong(int index);
public final long getLongLittleEndian(int index);
public final long getLongBigEndian(int index);
public final void putLong(int index, long value);
public final void putLongLittleEndian(int index, long value);
public final void putLongBigEndian(int index, long value);
public final float getFloat(int index);
public final float getFloatLittleEndian(int index);
public final float getFloatBigEndian(int index);
public final void putFloat(int index, float value);
public final void putFloatLittleEndian(int index, float value);
public final void putFloatBigEndian(int index, float value);
public final double getDouble(int index);
public final double getDoubleLittleEndian(int index);
public final double getDoubleBigEndian(int index);
public final void putDouble(int index, double value);
public final void putDoubleLittleEndian(int index, double value);
public final void putDoubleBigEndian(int index, double value);
public abstract void get(DataOutput out, int offset, int length) throws IOException;
public abstract void put(DataInput in, int offset, int length) throws IOException;
public abstract void get(int offset, ByteBuffer target, int numBytes);
public abstract void put(int offset, ByteBuffer source, int numBytes);
// 拷贝数据到目标segment
public final void copyTo(int offset, MemorySegment target, int targetOffset, int numBytes);
// 比较两个segment的数据
public final int compare(MemorySegment seg2, int offset1, int offset2, int len);
// 交互两个segment的数据
public final void swapBytes(byte[] tempBuffer, MemorySegment seg2, int offset1, int offset2, int len);
4. 总结
通过MemoryManager、MemoryPool、MemorySegment等类，Flink实现了应用层级对于内存的管理，规避了JVM原生内存管理带来的诸多问题，有效的提升了Flink的内存效率和性能。
