深度解析 Flink 是如何管理好内存的？

前言

如今，许多用于分析大型数据集的开源系统都是用 Java 或者是基于 JVM 的编程语言实现的。最着名的例子是 Apache Hadoop，还有较新的框架，如 Apache Spark、Apache Drill、Apache Flink。基于 JVM 的数据分析引擎面临的一个常见挑战就是如何在内存中存储大量的数据(包括缓存和高效处理)。合理的管理好 JVM 内存可以将 难以配置且不可预测的系统 与 少量配置且稳定运行的系统区分开来。

在这篇文章中，我们将讨论 Apache Flink 如何管理内存，讨论其自定义序列化与反序列化机制，以及它是如何操作二进制数据的。

数据对象直接放在堆内存中

在 JVM 中处理大量数据最直接的方式就是将这些数据做为对象存储在堆内存中，然后直接在内存中操作这些数据，如果想进行排序则就是对对象列表进行排序。然而这种方法有一些明显的缺点，首先，在频繁的创建和销毁大量对象的时候，监视和控制堆内存的使用并不是一件很简单的事情。如果对象分配过多的话，那么会导致内存过度使用，从而触发 OutOfMemoryError，导致 JVM 进程直接被杀死。另一个方面就是因为这些对象大都是生存在新生代，当 JVM 进行垃圾回收时，垃圾收集的开销很容易达到 50% 甚至更多。最后就是 Java 对象具有一定的空间开销(具体取决于 JVM 和平台)。对于具有许多小对象的数据集，这可以显著减少有效可用的内存量。如果你精通系统设计和系统调优，你可以根据系统进行特定的参数调整，可以或多或少的控制出现 OutOfMemoryError 的次数和避免堆内存的过多使用，但是这种设置和调优的作用有限，尤其是在数据量较大和执行环境发生变化的情况下。

Flink 是怎么做的?

Apache Flink 起源于一个研究项目，该项目旨在结合基于 MapReduce 的系统和并行数据库系统的最佳技术。在此背景下，Flink 一直有自己的内存数据处理方法。Flink 将对象序列化为固定数量的预先分配的内存段，而不是直接把对象放在堆内存上。它的 DBMS 风格的排序和连接算法尽可能多地对这个二进制数据进行操作，以此将序列化和反序列化开销降到最低。如果需要处理的数据多于可以保存在内存中的数据，Flink 的运算符会将部分数据溢出到磁盘。事实上，很多Flink 的内部实现看起来更像是 C / C ++，而不是普通的 Java。下图概述了 Flink 如何在内存段中存储序列化数据并在必要时溢出到磁盘：

Flink 的主动内存管理和操作二进制数据有几个好处：

1. 内存安全执行和高效的核外算法 由于分配的内存段的数量是固定的，因此监控剩余的内存资源是非常简单的。在内存不足的情况下，处理操作符可以有效地将更大批的内存段写入磁盘，后面再将它们读回到内存。因此，OutOfMemoryError 就有效的防止了。
2. 减少垃圾收集压力 因为所有长生命周期的数据都是在 Flink 的管理内存中以二进制表示的，所以所有数据对象都是短暂的，甚至是可变的，并且可以重用。短生命周期的对象可以更有效地进行垃圾收集，这大大降低了垃圾收集的压力。现在，预先分配的内存段是 JVM 堆上的长期存在的对象，为了降低垃圾收集的压力，Flink 社区正在积极地将其分配到堆外内存。这种努力将使得 JVM 堆变得更小，垃圾收集所消耗的时间将更少。
3. 节省空间的数据存储 Java 对象具有存储开销，如果数据以二进制的形式存储，则可以避免这种开销。
4. 高效的二进制操作和缓存敏感性 在给定合适的二进制表示的情况下，可以有效地比较和操作二进制数据。此外，二进制表示可以将相关值、哈希码、键和指针等相邻地存储在内存中。这使得数据结构通常具有更高效的缓存访问模式。

主动内存管理的这些特性在用于大规模数据分析的数据处理系统中是非常可取的，但是要实现这些功能的代价也是高昂的。要实现对二进制数据的自动内存管理和操作并非易事，使用 java.util.HashMap 比实现一个可溢出的 hash-table (由字节数组和自定义序列化支持)。当然，Apache Flink 并不是唯一一个基于 JVM 且对二进制数据进行操作的数据处理系统。例如 Apache Drill、Apache Ignite、Apache Geode 也有应用类似技术，最近 Apache Spark 也宣布将向这个方向演进。

下面我们将详细讨论 Flink 如何分配内存、如果对对象进行序列化和反序列化以及如果对二进制数据进行操作。我们还将通过一些性能表现数据来比较处理堆内存上的对象和对二进制数据的操作。

Flink 如何分配内存?

Flink TaskManager 是由几个内部组件组成的：actor 系统(负责与 Flink master 协调)、IOManager(负责将数据溢出到磁盘并将其读取回来)、MemoryManager(负责协调内存使用)。在本篇文章中，我们主要讲解 MemoryManager。

MemoryManager 负责将 MemorySegments 分配、计算和分发给数据处理操作符，例如 sort 和 join 等操作符。MemorySegment 是 Flink 的内存分配单元，由常规 Java 字节数组支持(默认大小为 32 KB)。MemorySegment 通过使用 Java 的 unsafe 方法对其支持的字节数组提供非常有效的读写访问。你可以将 MemorySegment 看作是 Java 的 NIO ByteBuffer 的定制版本。为了在更大的连续内存块上操作多个 MemorySegment，Flink 使用了实现 Java 的 java.io.DataOutput 和 java.io.DataInput 接口的逻辑视图。

MemorySegments 在 TaskManager 启动时分配一次，并在 TaskManager 关闭时销毁。因此，在 TaskManager 的整个生命周期中，MemorySegment 是重用的，而不会被垃圾收集的。在初始化 TaskManager 的所有内部数据结构并且已启动所有核心服务之后，MemoryManager 开始创建 MemorySegments。默认情况下，服务初始化后，70% 可用的 JVM 堆内存由 MemoryManager 分配(也可以配置全部)。剩余的 JVM 堆内存用于在任务处理期间实例化的对象，包括由用户定义的函数创建的对象。下图显示了启动后 TaskManager JVM 中的内存分布：

Flink 如何序列化对象?

（编辑：江门站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!