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

与其他的数据处理框架的 API(包括 SQL)类似，Flink 的 API 也提供了对数据集进行分组、排序和连接等转换操作。这些转换操作的数据集可能非常大。关系数据库系统具有非常高效的算法，比如 merge-sort、merge-join 和 hash-join。Flink 建立在这种技术的基础上，但是主要分为使用自定义序列化和自定义比较器来处理任意对象。在下面文章中我们将通过 Flink 的内存排序算法示例演示 Flink 如何使用二进制数据进行操作。

Flink 为其数据处理操作符预先分配内存，初始化时，排序算法从 MemoryManager 请求内存预算，并接收一组相应的 MemorySegments。这些 MemorySegments 变成了缓冲区的内存池，缓冲区中收集要排序的数据。下图说明了如何将数据对象序列化到排序缓冲区中：

排序缓冲区在内部分为两个内存区域：第一个区域保存所有对象的完整二进制数据，第二个区域包含指向完整二进制对象数据的指针(取决于 key 的数据类型)。将对象添加到排序缓冲区时，它的二进制数据会追加到第一个区域，指针(可能还有一个 key)被追加到第二个区域。分离实际数据和指针以及固定长度的 key 有两个目的：它可以有效的交换固定长度的 entries(key 和指针)，还可以减少排序时需要移动的数据。如果排序的 key 是可变长度的数据类型(比如 String)，则固定长度的排序 key 必须是前缀 key，比如字符串的前 n 个字符。请注意：并非所有数据类型都提供固定长度的前缀排序 key。将对象序列化到排序缓冲区时，两个内存区域都使用内存池中的 MemorySegments 进行扩展。一旦内存池为空且不能再添加对象时，则排序缓冲区将会被完全填充并可以进行排序。Flink 的排序缓冲区提供了比较和交换元素的方法，这使得实际的排序算法是可插拔的。默认情况下， Flink 使用了 Quicksort(快速排序)实现，可以使用 HeapSort(堆排序)。下图显示了如何比较两个对象：

排序缓冲区通过比较它们的二进制固定长度排序 key 来比较两个元素。如果元素的完整 key(不是前缀 key) 或者二进制前缀 key 不相等，则代表比较成功。如果前缀 key 相等(或者排序 key 的数据类型不提供二进制前缀 key)，则排序缓冲区遵循指向实际对象数据的指针，对两个对象进行反序列化并比较对象。根据比较结果，排序算法决定是否交换比较的元素。排序缓冲区通过移动其固定长度 key 和指针来交换两个元素，实际数据不会移动，排序算法完成后，排序缓冲区中的指针被正确排序。下图演示了如何从排序缓冲区返回已排序的数据：

通过顺序读取排序缓冲区的指针区域，跳过排序 key 并按照实际数据的排序指针返回排序数据。此数据要么反序列化并作为对象返回，要么在外部合并排序的情况下复制二进制数据并将其写入磁盘。

基准测试数据

（编辑：西安站长网）

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