深度解析 Flink 是如何管理好內(nèi)存的？
前言
如今，許多用于分析大型數(shù)據(jù)集的開源系統(tǒng)都是用 Java 或者是基于 JVM 的編程語言實現(xiàn)的。最著名的例子是 Apache Hadoop，還有較新的框架，如 Apache Spark、Apache Drill、Apache Flink。基于 JVM 的數(shù)據(jù)分析引擎面臨的一個常見挑戰(zhàn)就是如何在內(nèi)存中存儲大量的數(shù)據(jù)(包括緩存和高效處理)。合理的管理好 JVM 內(nèi)存可以將 難以配置且不可預(yù)測的系統(tǒng) 與 少量配置且穩(wěn)定運行的系統(tǒng)區(qū)分開來。
在這篇文章中，我們將討論 Apache Flink 如何管理內(nèi)存，討論其自定義序列化與反序列化機制，以及它是如何操作二進(jìn)制數(shù)據(jù)的。
數(shù)據(jù)對象直接放在堆內(nèi)存中
在 JVM 中處理大量數(shù)據(jù)最直接的方式就是將這些數(shù)據(jù)做為對象存儲在堆內(nèi)存中，然后直接在內(nèi)存中操作這些數(shù)據(jù)，如果想進(jìn)行排序則就是對對象列表進(jìn)行排序。然而這種方法有一些明顯的缺點，首先，在頻繁的創(chuàng)建和銷毀大量對象的時候，監(jiān)視和控制堆內(nèi)存的使用并不是一件很簡單的事情。如果對象分配過多的話，那么會導(dǎo)致內(nèi)存過度使用，從而觸發(fā) OutOfMemoryError，導(dǎo)致 JVM 進(jìn)程直接被殺死。另一個方面就是因為這些對象大都是生存在新生代，當(dāng) JVM 進(jìn)行垃圾回收時，垃圾收集的開銷很容易達(dá)到 50% 甚至更多。最后就是 Java 對象具有一定的空間開銷(具體取決于 JVM 和平臺)。對于具有許多小對象的數(shù)據(jù)集，這可以顯著減少有效可用的內(nèi)存量。如果你精通系統(tǒng)設(shè)計和系統(tǒng)調(diào)優(yōu)，你可以根據(jù)系統(tǒng)進(jìn)行特定的參數(shù)調(diào)整，可以或多或少的控制出現(xiàn) OutOfMemoryError 的次數(shù)和避免堆內(nèi)存的過多使用，但是這種設(shè)置和調(diào)優(yōu)的作用有限，尤其是在數(shù)據(jù)量較大和執(zhí)行環(huán)境發(fā)生變化的情況下。
Flink 是怎么做的?
Apache Flink 起源于一個研究項目，該項目旨在結(jié)合基于 MapReduce 的系統(tǒng)和并行數(shù)據(jù)庫系統(tǒng)的最佳技術(shù)。在此背景下，F(xiàn)link 一直有自己的內(nèi)存數(shù)據(jù)處理方法。Flink 將對象序列化為固定數(shù)量的預(yù)先分配的內(nèi)存段，而不是直接把對象放在堆內(nèi)存上。它的 DBMS 風(fēng)格的排序和連接算法盡可能多地對這個二進(jìn)制數(shù)據(jù)進(jìn)行操作，以此將序列化和反序列化開銷降到最低。如果需要處理的數(shù)據(jù)多于可以保存在內(nèi)存中的數(shù)據(jù)，F(xiàn)link 的運算符會將部分?jǐn)?shù)據(jù)溢出到磁盤。事實上，很多Flink 的內(nèi)部實現(xiàn)看起來更像是 C / C ++，而不是普通的 Java。下圖概述了 Flink 如何在內(nèi)存段中存儲序列化數(shù)據(jù)并在必要時溢出到磁盤：
Flink 的主動內(nèi)存管理和操作二進(jìn)制數(shù)據(jù)有幾個好處：
- 內(nèi)存安全執(zhí)行和高效的核外算法 由于分配的內(nèi)存段的數(shù)量是固定的，因此監(jiān)控剩余的內(nèi)存資源是非常簡單的。在內(nèi)存不足的情況下，處理操作符可以有效地將更大批的內(nèi)存段寫入磁盤，后面再將它們讀回到內(nèi)存。因此，OutOfMemoryError 就有效的防止了。
- 減少垃圾收集壓力 因為所有長生命周期的數(shù)據(jù)都是在 Flink 的管理內(nèi)存中以二進(jìn)制表示的，所以所有數(shù)據(jù)對象都是短暫的，甚至是可變的，并且可以重用。短生命周期的對象可以更有效地進(jìn)行垃圾收集，這大大降低了垃圾收集的壓力。現(xiàn)在，預(yù)先分配的內(nèi)存段是 JVM 堆上的長期存在的對象，為了降低垃圾收集的壓力，F(xiàn)link 社區(qū)正在積極地將其分配到堆外內(nèi)存。這種努力將使得 JVM 堆變得更小，垃圾收集所消耗的時間將更少。
- 節(jié)省空間的數(shù)據(jù)存儲 Java 對象具有存儲開銷，如果數(shù)據(jù)以二進(jìn)制的形式存儲，則可以避免這種開銷。
- 高效的二進(jìn)制操作和緩存敏感性 在給定合適的二進(jìn)制表示的情況下，可以有效地比較和操作二進(jìn)制數(shù)據(jù)。此外，二進(jìn)制表示可以將相關(guān)值、哈希碼、鍵和指針等相鄰地存儲在內(nèi)存中。這使得數(shù)據(jù)結(jié)構(gòu)通常具有更高效的緩存訪問模式。
主動內(nèi)存管理的這些特性在用于大規(guī)模數(shù)據(jù)分析的數(shù)據(jù)處理系統(tǒng)中是非常可取的，但是要實現(xiàn)這些功能的代價也是高昂的。要實現(xiàn)對二進(jìn)制數(shù)據(jù)的自動內(nèi)存管理和操作并非易事，使用 java.util.HashMap 比實現(xiàn)一個可溢出的 hash-table (由字節(jié)數(shù)組和自定義序列化支持)。當(dāng)然，Apache Flink 并不是唯一一個基于 JVM 且對二進(jìn)制數(shù)據(jù)進(jìn)行操作的數(shù)據(jù)處理系統(tǒng)。例如 Apache Drill、Apache Ignite、Apache Geode 也有應(yīng)用類似技術(shù)，最近 Apache Spark 也宣布將向這個方向演進(jìn)。
下面我們將詳細(xì)討論 Flink 如何分配內(nèi)存、如果對對象進(jìn)行序列化和反序列化以及如果對二進(jìn)制數(shù)據(jù)進(jìn)行操作。我們還將通過一些性能表現(xiàn)數(shù)據(jù)來比較處理堆內(nèi)存上的對象和對二進(jìn)制數(shù)據(jù)的操作。
Flink 如何分配內(nèi)存?
Flink TaskManager 是由幾個內(nèi)部組件組成的：actor 系統(tǒng)(負(fù)責(zé)與 Flink master 協(xié)調(diào))、IOManager(負(fù)責(zé)將數(shù)據(jù)溢出到磁盤并將其讀取回來)、MemoryManager(負(fù)責(zé)協(xié)調(diào)內(nèi)存使用)。在本篇文章中，我們主要講解 MemoryManager。
MemoryManager 負(fù)責(zé)將 MemorySegments 分配、計算和分發(fā)給數(shù)據(jù)處理操作符，例如 sort 和 join 等操作符。MemorySegment 是 Flink 的內(nèi)存分配單元，由常規(guī) Java 字節(jié)數(shù)組支持(默認(rèn)大小為 32 KB)。MemorySegment 通過使用 Java 的 unsafe 方法對其支持的字節(jié)數(shù)組提供非常有效的讀寫訪問。你可以將 MemorySegment 看作是 Java 的 NIO ByteBuffer 的定制版本。為了在更大的連續(xù)內(nèi)存塊上操作多個 MemorySegment，F(xiàn)link 使用了實現(xiàn) Java 的 java.io.DataOutput 和 java.io.DataInput 接口的邏輯視圖。
MemorySegments 在 TaskManager 啟動時分配一次，并在 TaskManager 關(guān)閉時銷毀。因此，在 TaskManager 的整個生命周期中，MemorySegment 是重用的，而不會被垃圾收集的。在初始化 TaskManager 的所有內(nèi)部數(shù)據(jù)結(jié)構(gòu)并且已啟動所有核心服務(wù)之后，MemoryManager 開始創(chuàng)建 MemorySegments。默認(rèn)情況下，服務(wù)初始化后，70% 可用的 JVM 堆內(nèi)存由 MemoryManager 分配(也可以配置全部)。剩余的 JVM 堆內(nèi)存用于在任務(wù)處理期間實例化的對象，包括由用戶定義的函數(shù)創(chuàng)建的對象。下圖顯示了啟動后 TaskManager JVM 中的內(nèi)存分布：
Flink 如何序列化對象?
Java 生態(tài)系統(tǒng)提供了幾個庫，可以將對象轉(zhuǎn)換為二進(jìn)制表示形式并返回。常見的替代方案是標(biāo)準(zhǔn) Java 序列化，Kryo，Apache Avro，Apache Thrift 或 Google 的 Protobuf。Flink 包含自己的自定義序列化框架，以便控制數(shù)據(jù)的二進(jìn)制表示。這一點很重要，因為對二進(jìn)制數(shù)據(jù)進(jìn)行操作需要對序列化布局有準(zhǔn)確的了解。此外，根據(jù)在二進(jìn)制數(shù)據(jù)上執(zhí)行的操作配置序列化布局可以顯著提升性能。Flink 的序列化機制利用了這一特性，即在執(zhí)行程序之前，要序列化和反序列化的對象的類型是完全已知的。
Flink 程序可以處理表示為任意 Java 或 Scala 對象的數(shù)據(jù)。在優(yōu)化程序之前，需要識別程序數(shù)據(jù)流的每個處理步驟中的數(shù)據(jù)類型。對于 Java 程序，F(xiàn)link 提供了一個基于反射的類型提取組件，用于分析用戶定義函數(shù)的返回類型。Scala 程序可以在 Scala 編譯器的幫助下進(jìn)行分析。Flink 使用 TypeInformation 表示每種數(shù)據(jù)類型。
- Flink 有如下幾種數(shù)據(jù)類型的 TypeInformations：
- BasicTypeInfo：所有 Java 的基礎(chǔ)類型或 java.lang.String
- BasicArrayTypeInfo：Java 基本類型構(gòu)成的數(shù)組或 java.lang.String
- WritableTypeInfo：Hadoop 的 Writable 接口的任何實現(xiàn)
- TupleTypeInfo：任何 Flink tuple(Tuple1 到 Tuple25)。Flink tuples 是具有類型化字段的固定長度元組的 Java 表示
- CaseClassTypeInfo：任何 Scala CaseClass(包括 Scala tuples)
- PojoTypeInfo：任何 POJO(Java 或 Scala)，即所有字段都是 public 的或通過 getter 和 setter 訪問的對象，遵循通用命名約定
- GenericTypeInfo：不能標(biāo)識為其他類型的任何數(shù)據(jù)類型
每個 TypeInformation 都為它所代表的數(shù)據(jù)類型提供了一個序列化器。例如，BasicTypeInfo 返回一個序列化器，該序列化器寫入相應(yīng)的基本類型;WritableTypeInfo 的序列化器將序列化和反序列化委托給實現(xiàn) Hadoop 的 Writable 接口的對象的 write() 和 readFields() 方法;GenericTypeInfo 返回一個序列化器，該序列化器將序列化委托給 Kryo。對象將自動通過 Java 中高效的 Unsafe 方法來序列化到 Flink MemorySegments 支持的 DataOutput。對于可用作鍵的數(shù)據(jù)類型，例如哈希值，TypeInformation 提供了 TypeComparators，TypeComparators 比較和哈希對象，并且可以根據(jù)具體的數(shù)據(jù)類型有效的比較二進(jìn)制并提取固定長度的二進(jìn)制 key 前綴。
Tuple，Pojo 和 CaseClass 類型是復(fù)合類型，它們可能嵌套一個或者多個數(shù)據(jù)類型。因此，它們的序列化和比較也都比較復(fù)雜，一般將其成員數(shù)據(jù)類型的序列化和比較都交給各自的 Serializers(序列化器) 和 Comparators(比較器)。下圖說明了 Tuple3對象的序列化，其中Person 是 POJO 并定義如下：
- public class Person {
- public int id;
- public String name;
- }
通過提供定制的 TypeInformations、Serializers(序列化器) 和 Comparators(比較器)，可以方便地擴展 Flink 的類型系統(tǒng)，從而提高序列化和比較自定義數(shù)據(jù)類型的性能。
Flink 如何對二進(jìn)制數(shù)據(jù)進(jìn)行操作?
與其他的數(shù)據(jù)處理框架的 API(包括 SQL)類似，F(xiàn)link 的 API 也提供了對數(shù)據(jù)集進(jìn)行分組、排序和連接等轉(zhuǎn)換操作。這些轉(zhuǎn)換操作的數(shù)據(jù)集可能非常大。關(guān)系數(shù)據(jù)庫系統(tǒng)具有非常高效的算法，比如 merge-sort、merge-join 和 hash-join。Flink 建立在這種技術(shù)的基礎(chǔ)上，但是主要分為使用自定義序列化和自定義比較器來處理任意對象。在下面文章中我們將通過 Flink 的內(nèi)存排序算法示例演示 Flink 如何使用二進(jìn)制數(shù)據(jù)進(jìn)行操作。
Flink 為其數(shù)據(jù)處理操作符預(yù)先分配內(nèi)存，初始化時，排序算法從 MemoryManager 請求內(nèi)存預(yù)算，并接收一組相應(yīng)的 MemorySegments。這些 MemorySegments 變成了緩沖區(qū)的內(nèi)存池，緩沖區(qū)中收集要排序的數(shù)據(jù)。下圖說明了如何將數(shù)據(jù)對象序列化到排序緩沖區(qū)中：
排序緩沖區(qū)在內(nèi)部分為兩個內(nèi)存區(qū)域：第一個區(qū)域保存所有對象的完整二進(jìn)制數(shù)據(jù)，第二個區(qū)域包含指向完整二進(jìn)制對象數(shù)據(jù)的指針(取決于 key 的數(shù)據(jù)類型)。將對象添加到排序緩沖區(qū)時，它的二進(jìn)制數(shù)據(jù)會追加到第一個區(qū)域，指針(可能還有一個 key)被追加到第二個區(qū)域。分離實際數(shù)據(jù)和指針以及固定長度的 key 有兩個目的：它可以有效的交換固定長度的 entries(key 和指針)，還可以減少排序時需要移動的數(shù)據(jù)。如果排序的 key 是可變長度的數(shù)據(jù)類型(比如 String)，則固定長度的排序 key 必須是前綴 key，比如字符串的前 n 個字符。請注意：并非所有數(shù)據(jù)類型都提供固定長度的前綴排序 key。將對象序列化到排序緩沖區(qū)時，兩個內(nèi)存區(qū)域都使用內(nèi)存池中的 MemorySegments 進(jìn)行擴展。一旦內(nèi)存池為空且不能再添加對象時，則排序緩沖區(qū)將會被完全填充并可以進(jìn)行排序。Flink 的排序緩沖區(qū)提供了比較和交換元素的方法，這使得實際的排序算法是可插拔的。默認(rèn)情況下， Flink 使用了 Quicksort(快速排序)實現(xiàn)，可以使用 HeapSort(堆排序)。下圖顯示了如何比較兩個對象：
排序緩沖區(qū)通過比較它們的二進(jìn)制固定長度排序 key 來比較兩個元素。如果元素的完整 key(不是前綴 key) 或者二進(jìn)制前綴 key 不相等，則代表比較成功。如果前綴 key 相等(或者排序 key 的數(shù)據(jù)類型不提供二進(jìn)制前綴 key)，則排序緩沖區(qū)遵循指向?qū)嶋H對象數(shù)據(jù)的指針，對兩個對象進(jìn)行反序列化并比較對象。根據(jù)比較結(jié)果，排序算法決定是否交換比較的元素。排序緩沖區(qū)通過移動其固定長度 key 和指針來交換兩個元素，實際數(shù)據(jù)不會移動，排序算法完成后，排序緩沖區(qū)中的指針被正確排序。下圖演示了如何從排序緩沖區(qū)返回已排序的數(shù)據(jù)：
通過順序讀取排序緩沖區(qū)的指針區(qū)域，跳過排序 key 并按照實際數(shù)據(jù)的排序指針返回排序數(shù)據(jù)。此數(shù)據(jù)要么反序列化并作為對象返回，要么在外部合并排序的情況下復(fù)制二進(jìn)制數(shù)據(jù)并將其寫入磁盤。
基準(zhǔn)測試數(shù)據(jù)
那么，對二進(jìn)制數(shù)據(jù)進(jìn)行操作對性能意味著什么?我們將運行一個基準(zhǔn)測試，對 1000 萬個Tuple2對象進(jìn)行排序以找出答案。整數(shù)字段的值從均勻分布中采樣。String 字段值的長度為 12 個字符，并從長尾分布中進(jìn)行采樣。輸入數(shù)據(jù)由返回可變對象的迭代器提供，即返回具有不同字段值的相同 Tuple 對象實例。Flink 在從內(nèi)存，網(wǎng)絡(luò)或磁盤讀取數(shù)據(jù)時使用此技術(shù)，以避免不必要的對象實例化。基準(zhǔn)測試在具有 900 MB 堆大小的 JVM 中運行，在堆上存儲和排序 1000 萬個 Tuple 對象并且不會導(dǎo)致觸發(fā) OutOfMemoryError 大約需要這么大的內(nèi)存。我們使用三種排序方法在Integer 字段和 String 字段上對 Tuple 對象進(jìn)行排序：
- 對象存在堆中：Tuple 對象存儲在常用的 java.util.ArrayList 中，初始容量設(shè)置為 1000 萬，并使用 Java 中常用的集合排序進(jìn)行排序。
- Flink 序列化：使用 Flink 的自定義序列化程序?qū)?Tuple 字段序列化為 600 MB 大小的排序緩沖區(qū)，如上所述排序，最后再次反序列化。在 Integer 字段上進(jìn)行排序時，完整的 Integer 用作排序 key，以便排序完全發(fā)生在二進(jìn)制數(shù)據(jù)上(不需要對象的反序列化)。對于 String 字段的排序，使用 8 字節(jié)前綴 key，如果前綴 key 相等，則對 Tuple 對象進(jìn)行反序列化。
- Kryo 序列化：使用 Kryo 序列化將 Tuple 字段序列化為 600 MB 大小的排序緩沖區(qū)，并在沒有二進(jìn)制排序 key 的情況下進(jìn)行排序。這意味著每次比較需要對兩個對象進(jìn)行反序列化。
所有排序方法都使用單線程實現(xiàn)。結(jié)果的時間是十次運行結(jié)果的平均值。在每次運行之后，我們調(diào)用System.gc()請求垃圾收集運行，該運行不會進(jìn)入測量的執(zhí)行時間。下圖顯示了將輸入數(shù)據(jù)存儲在內(nèi)存中，對其進(jìn)行排序并將其作為對象讀回的時間。
我們看到 Flink 使用自己的序列化器對二進(jìn)制數(shù)據(jù)進(jìn)行排序明顯優(yōu)于其他兩種方法。與存儲在堆內(nèi)存上相比，我們看到將數(shù)據(jù)加載到內(nèi)存中要快得多。因為我們實際上是在收集對象，沒有機會重用對象實例，但必須重新創(chuàng)建每個 Tuple。這比 Flink 的序列化器(或Kryo序列化)效率低。另一方面，與反序列化相比，從堆中讀取對象是無性能消耗的。在我們的基準(zhǔn)測試中，對象克隆比序列化和反序列化組合更耗性能。查看排序時間，我們看到對二進(jìn)制數(shù)據(jù)的排序也比 Java 的集合排序更快。使用沒有二進(jìn)制排序 key 的 Kryo 序列化的數(shù)據(jù)排序比其他方法慢得多。這是因為反序列化帶來很大的開銷。在String 字段上對 Tuple 進(jìn)行排序比在 Integer 字段上排序更快，因為長尾值分布顯著減少了成對比較的數(shù)量。為了更好地了解排序過程中發(fā)生的狀況，我們使用 VisualVM 監(jiān)控執(zhí)行的 JVM。以下截圖顯示了執(zhí)行 10次 運行時的堆內(nèi)存使用情況、垃圾收集情況和 CPU 使用情況。
測試是在 8 核機器上運行單線程，因此一個核心的完全利用僅對應(yīng) 12.5% 的總體利用率。截圖顯示，對二進(jìn)制數(shù)據(jù)進(jìn)行操作可顯著減少垃圾回收活動。對于對象存在堆中，垃圾收集器在排序緩沖區(qū)被填滿時以非常短的時間間隔運行，并且即使對于單個處理線程也會導(dǎo)致大量 CPU 使用(排序本身不會觸發(fā)垃圾收集器)。JVM 垃圾收集多個并行線程，解釋了高CPU 總體利用率。另一方面，對序列化數(shù)據(jù)進(jìn)行操作的方法很少觸發(fā)垃圾收集器并且 CPU 利用率低得多。實際上，如果使用 Flink 序列化的方式在 Integer 字段上對 Tuple 進(jìn)行排序，則垃圾收集器根本不運行，因為對于成對比較，不需要反序列化任何對象。Kryo 序列化需要比較多的垃圾收集，因為它不使用二進(jìn)制排序 key 并且每次排序都要反序列化兩個對象。
內(nèi)存使用情況上圖顯示 Flink 序列化和 Kryo 序列化不斷的占用大量內(nèi)存
存使用情況圖表顯示flink-serialized和kryo-serialized不斷占用大量內(nèi)存。這是由于 MemorySegments 的預(yù)分配。實際內(nèi)存使用率要低得多，因為排序緩沖區(qū)并未完全填充。下表顯示了每種方法的內(nèi)存消耗。1000 萬條數(shù)據(jù)產(chǎn)生大約 280 MB 的二進(jìn)制數(shù)據(jù)(對象數(shù)據(jù)、指針和排序 key)，具體取決于使用的序列化程序以及二進(jìn)制排序 key 的存在和大小。將其與數(shù)據(jù)存儲在堆上的方法進(jìn)行比較，我們發(fā)現(xiàn)對二進(jìn)制數(shù)據(jù)進(jìn)行操作可以顯著提高內(nèi)存效率。在我們的基準(zhǔn)測試中，如果序列化為排序緩沖區(qū)而不是將其作為堆上的對象保存，則可以在內(nèi)存中對兩倍以上的數(shù)據(jù)進(jìn)行排序。
總而言之，測試驗證了文章前面說的對二進(jìn)制數(shù)據(jù)進(jìn)行操作的好處。
展望未來
Apache Flink 具有相當(dāng)多的高級技術(shù)，可以通過有限的內(nèi)存資源安全有效地處理大量數(shù)據(jù)。但是有幾點可以使 Flink 更有效率。Flink 社區(qū)正在努力將管理內(nèi)存移動到堆外內(nèi)存。這將允許更小的 JVM，更低的垃圾收集開銷，以及更容易的系統(tǒng)配置。使用 Flink 的 Table API，所有操作(如 aggregation 和 projection)的語義都是已知的(與黑盒用戶定義的函數(shù)相反)。因此，我們可以為直接對二進(jìn)制數(shù)據(jù)進(jìn)行操作的 Table API 操作生成代碼。進(jìn)一步的改進(jìn)包括序列化設(shè)計，這些設(shè)計針對應(yīng)用于二進(jìn)制數(shù)據(jù)的操作和針對序列化器和比較器的代碼生成而定制。
總結(jié)
- Flink 的主動內(nèi)存管理減少了因觸發(fā) OutOfMemoryErrors 而殺死 JVM 進(jìn)程和垃圾收集開銷的問題。
- Flink 具有高效的數(shù)據(jù)序列化和反序列化機制，有助于對二進(jìn)制數(shù)據(jù)進(jìn)行操作，并使更多數(shù)據(jù)適合內(nèi)存。
- Flink 的 DBMS 風(fēng)格的運算符本身在二進(jìn)制數(shù)據(jù)上運行，在必要時可以在內(nèi)存中高性能地傳輸?shù)酱疟P。
