6. Flink 的 kafka 聯結器有什麼特別的地方？

Flink原始碼中有一個獨立的connectors模組，所有的其他connector都依賴於此模組，Flink 在1.9版本釋出的全新kafka聯結器，摒棄了之前連線不同版本的kafka叢集需要依賴不同版本的connector這種做法，只需要依賴一個connector即可。

7. 說說 Flink的記憶體管理是如何做的?

Flink 並不是將大量物件存在堆上，而是將物件都序列化到一個預分配的記憶體塊上。

此外，Flink大量的使用了堆外記憶體。如果需要處理的資料超出了記憶體限制，則會將部分資料儲存到硬碟上。

Flink 為了直接操作二進位制資料實現了自己的序列化框架。

理論上Flink的記憶體管理分為三部分：

Network Buffers：這個是在TaskManager啟動的時候分配的，這是一組用於快取網路資料的記憶體，每個塊是32K，預設分配2048個，可以通過“taskmanager.network.numberOfBuffers”修改

Memory Manage pool：大量的Memory Segment塊，用於執行時的演算法（Sort/Join/Shuffle等），這部分啟動的時候就會分配。下面這段程式碼，根據配置檔案中的各種引數來計算記憶體的分配方法。（heap or off-heap，這個放到下節談），記憶體的分配支援預分配和lazy load，預設懶載入的方式。

User Code，這部分是除了Memory Manager之外的記憶體用於User code和TaskManager本身的資料結構。

8. 說說 Flink的序列化如何做的?

Java本身自帶的序列化和反序列化的功能，但是輔助資訊佔用空間比較大，在序列化物件時記錄了過多的類資訊。Apache Flink摒棄了Java原生的序列化方法，以獨特的方式處理資料型別和序列化，包含自己的型別描述符，泛型型別提取和型別序列化框架。TypeInformation 是所有型別描述符的基類。它揭示了該型別的一些基本屬性，並且可以生成序列化器。TypeInformation 支援以下幾種型別：

BasicTypeInfo: 任意Java 基本型別或 String 型別

BasicArrayTypeInfo: 任意Java基本型別陣列或 String 陣列

WritableTypeInfo: 任意 Hadoop Writable 介面的實現類

TupleTypeInfo: 任意的 Flink Tuple 型別(支援Tuple1 to Tuple25)。Flink tuples 是固定長度固定型別的Java Tuple實現

CaseClassTypeInfo: 任意的 Scala CaseClass(包括 Scala tuples)

PojoTypeInfo: 任意的 POJO (Java or Scala)，例如，Java物件的所有成員變數，要麼是 public 修飾符定義，要麼有 getter/setter 方法

GenericTypeInfo: 任意無法匹配之前幾種型別的類

針對前六種型別資料集，Flink皆可以自動生成對應的TypeSerializer，能非常高效地對資料集進行序列化和反序列化。

9. Flink中的Window出現了資料傾斜，你有什麼解決辦法？

window產生資料傾斜指的是資料在不同的視窗內堆積的資料量相差過多。本質上產生這種情況的原因是資料來源頭髮送的資料量速度不同導致的。出現這種情況一般通過兩種方式來解決：

在資料進入視窗前做預聚合

重新設計視窗聚合的key

10. Flink中在使用聚合函式 GroupBy、Distinct、KeyBy 等函式時出現數據熱點該如何解決？

資料傾斜和資料熱點是所有大資料框架繞不過去的問題。處理這類問題主要從3個方面入手：

在業務上規避這類問題

例如一個假設訂單場景，北京和上海兩個城市訂單量增長几十倍，其餘城市的資料量不變。這時候我們在進行聚合的時候，北京和上海就會出現資料堆積，我們可以單獨資料北京和上海的資料。

Key的設計上

把熱key進行拆分，比如上個例子中的北京和上海，可以把北京和上海按照地區進行拆分聚合。

引數設定

Flink 1.9.0 SQL(Blink Planner) 效能優化中一項重要的改進就是升級了微批模型，即 MiniBatch。原理是快取一定的資料後再觸發處理，以減少對State的訪問，從而提升吞吐和減少資料的輸出量。