二、flink的記憶體管理機制

0.flink的記憶體劃分

1.flink在JVM的heap內有自己的記憶體管理空間。
2.在flink中記憶體被分為三個部分，分別是Unmanaged區域，Managed區域，Network-Buffer區域
3.Unmanaged區域是指flink不管理這部分割槽域，它的管理由JVM管理，用於存放User Code
4.Managed區域是指flink管理這部分割槽域，它不受jvm管理不存在GC問題，用於存放Hashing,Sorting,Caching等資料
5.Network-Buffer區域是指flink在進行計算時需要通過網路進行交換資料的區域。用於存放Shuffles，Broadcasts資料。
 

1.flink使用堆外記憶體

1.為了解決大量物件在JVM的heap上建立會帶來OOM和GC的問題，flink將大量使用的記憶體存放到堆外.
2.flink在堆外有一塊預分配的固定大小的記憶體塊MemorySegment，flink會將物件高效的序列化到這塊記憶體中。
  MemorySegment由許多小的記憶體cell組成，每個cell大小32kb，這也是flink分配記憶體的最小單位。你可以把 
  MemorySegment想象成是為Flink 定製的 java.nio.ByteBuffer。它的底層可以是一個普通的 Java 位元組陣列
 （byte[]），也可以是一個申請在堆外的 ByteBuffer。每條記錄都會以序列化的形式儲存在一個或多個MemorySegment中。
3.
 
如果MemorySegment中依然放不小所有的資料，flink會將資料寫入磁碟，需要的時候再衝磁碟讀出來。

使用堆外記憶體獲得的好處：
1.有效防止OOM
由於MemorySegment大小固定，操作高效。如果MemorySegment不足寫出到磁碟，記憶體中的資料不多，一般不會發生OOM.

2.大幅度減輕GC壓力
少量長期使用的資料以二進位制形式儲存在記憶體，大量臨時使用的物件被序列化到磁碟。物件數量就會大量減少，GC壓力減輕。

3.節省記憶體空間
Java物件的儲存密度叫低，現在大量資料都是二進位制的表示形式，儲存密度提高了，記憶體利用率提高了！

4.二進位制操作更高效,快取操作更友好
二進位制資料以定義好的格式儲存，可以高效地比較與操作。另外，該二進位制形式可以把相關的值，以及hash
 
值，鍵值和指標等相鄰
地放進記憶體中。這使得資料結構可以對快取記憶體更友好，可以從 L1/L2/L3 快取獲得性能的提升

2.flink量身打造序列化方案

假設有一個Tuple3

//1.Person型別
public class Person {
    public int id;
    public String name;
}
//Tuple3<age:Integer, height:Double, Person>物件
(25,175.5,Person(1,"zhangsan"))

它將被序列化為下面的形式:

可見這種序列化方式儲存密度是相當緊湊的。其中int佔4位元組，double佔8位元組，POJO多個一個位元組的header，
PojoSerializer只負責將header序列化進去，並委託每個欄位對應的serializer對欄位進行序列化。

1.Java生態圈提供了許多序列化框架諸如Java serialization, Kryo, Apache Avro等等。
2.flink中處理的資料流通常是同一型別,物件的型別是固定，可以對整個資料流只儲存一份物件Schema資訊，這將大大節省儲存空間
3.由於型別固定，當我們需要訪問某個物件成員變數的時候，可以通過偏移量直接存取，並不需要反序列化整個Java物件。
4.Flink序列化框架支援任意的Java或是Scala型別，並且不需要像Hadoop那樣必須實現org.apache.hadoop.io.Writable介面。
5.Flink序列化框架支援資料型別自動識別。
  如果是Java程式，通過Java Reflection分析UDF (User Define Function)的返回值型別確定資料型別。
  如果是Scala程式，通過Scala Compiler分析UDF (User Define Function)的返回值型別確定資料型別。

型別資訊由 TypeInformation 類表示，TypeInformation 支援以下幾種型別：

BasicTypeInfo: 任意Java基本型別(裝箱的)或String型別。
BasicArrayTypeInfo: 任意Java基本型別陣列(裝箱的)或String型別。
WritableTypeInfo: 任意Hadoop Writable介面的實現類。
TupleTypeInfo: 任意的Flink Tuple 型別(支援Tuple1到Tuple25)。Flink tuples是固定長度固定型別的Java Tuple
CaseClassTypeInfo: 任意的 Scala CaseClass(包括 Scala tuples)。
PojoTypeInfo: 任意的POJO(Java或Scala)Java物件的所有成員變數，或public修飾符定義，或有getter/setter方法。
GenericTypeInfo: 任意無法匹配之前幾種型別的類。

注意：
1.前六種型別資料集，Flink皆可以自動生成對應的TypeSerializer，能非常高效地對資料集進行序列化和反序列化。
2.最後一種資料型別，Flink會使用Kryo進行序列化和反序列化。
3.對於 Tuple、CaseClass、POJO 等組合型別，其TypeSerializer和TypeComparator也是組合的，
  序列化和比較時會委託給對應的serializers和comparators

3.flink採用資料和引用分開儲存的方式操作資料

1.flink提供大量的api,有些sql-api或sort，group，join等操作牽涉到大量的資料，使用大量記憶體。
2.這些操作都是基於flink的資料記憶體和引用記憶體分開儲存的方式進行操作的。

以sort為例：

1.Flink從MemoryManager申請一批MemorySegment，作為sort-buffer，用來存放排序的資料。
2.sort-buffer分成兩塊
  一塊用來存放所有物件完整的二進位制資料。
  一塊用來存放指向完整二進位制資料的引用。
  引用由指標(pointer)定長的序列化後的鍵（key）組成，ref=point+key
      將key和point分開儲存的動機是：
      a.ref.point指向真實資料塊，
      b.ref.key用來做基於key的諸如compare等操作，
      c.ref.key是連續儲存的，這樣能提高cpu的快取命中率，加快CPU訪問資料。
3.當一個物件要加到 sort-buffer時，它的binary-data會被加到第一個區域，ref=(piont+key)會被加到第二個區域。
4.執行比較時，如果有binary-key直接通過偏移量操作binary-key.如果沒有binary-key那隻能序列化整個物件再進行比較。
5.執行互動時，只需互動ref,不需要互動binary-data.

6.訪問資料時，只需沿著排好序的ref區域順序訪問，通過ref.pointer找到對應的真實資料.