Flink 面試--進階篇
1、Flink是如何支持批流一體的？
2、Flink是如何做到高效的數據交換的？
3、Flink是如何做容錯的？
4、Flink 分布式快照的原理是什么？
5、Flink 是如何保證Exactly-once語義的？
6、Flink 的 kafka 連接器有什么特別的地方？
7、說說 Flink的內存管理是如何做的?
8、說說 Flink的序列化如何做的?
9、Flink中的Window出現了數據傾斜，你有什么解決辦法？
10、
Flink中在使用聚合函數 GroupBy、Distinct、KeyBy 等函數時出現數據熱點該如何解決？
11、Flink任務延遲高，想解決這個問題，你會如何入手？
12、Flink是如何處理反壓的？
13、Operator Chains（算子鏈）這個概念你了解嗎？
14、
Flink什么情況下才會把Operator chain在一起形成算子鏈？
一、Flink是如何支持批流一體的？
本道面試題考察的其實就是一句話：Flink的開發者認為批處理是流處理的一種特殊情況。批處理是有限的流處理。Flink 使用一個引擎支持了DataSet API 和 DataStream API。
二、Flink是如何做到高效的數據交換的？
在一個Flink Job中，數據需要在不同的task中進行交換，整個數據交換是有 TaskManager 負責的，TaskManager 的網絡組件首先從緩沖buffer中收集records，然后再發送。Records 並不是一個一個被發送的，而是積累一個批次再發送，batch 技術可以更加高效的利用網絡資源。
三、Flink是如何做容錯的？
Flink 實現容錯主要靠強大的CheckPoint機制和State機制。Checkpoint 負責定時制作分布式快照、對程序中的狀態進行備份；State 用來存儲計算過程中的中間狀態。
四、Flink 分布式快照的原理是什么？
Flink的分布式快照是根據Chandy-Lamport算法量身定做的。簡單來說就是持續創建分布式數據流及其狀態的一致快照。
核心思想是在 input source 端插入 barrier，控制 barrier 的同步來實現 snapshot 的備份和 exactly-once 語義。
五、Flink 是如何保證Exactly-once語義的？
Flink通過實現兩階段提交和狀態保存來實現端到端的一致性語義。分為以下幾個步驟：
開始事務（beginTransaction）創建一個臨時文件夾，來寫把數據寫入到這個文件夾里面
預提交（preCommit）將內存中緩存的數據寫入文件並關閉
正式提交（commit）將之前寫完的臨時文件放入目標目錄下。這代表着最終的數據會有一些延遲
丟棄（abort）丟棄臨時文件
若失敗發生在預提交成功后，正式提交前。可以根據狀態來提交預提交的數據，也可刪除預提交的數據。
七、說說 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本身的數據結構。
八、說說 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，能非常高效地對數據集進行序列化和反序列化。
九、 Flink中的Window出現了數據傾斜，你有什么解決辦法？
window產生數據傾斜指的是數據在不同的窗口內堆積的數據量相差過多。本質上產生這種情況的原因是數據源頭發送的數據量速度不同導致的。出現這種情況一般通過兩種方式來解決：
在數據進入窗口前做預聚合
重新設計窗口聚合的key
十、 Flink中在使用聚合函數 GroupBy、Distinct、KeyBy 等函數時出現數據熱點該如何解決？
數據傾斜和數據熱點是所有大數據框架繞不過去的問題。處理這類問題主要從3個方面入手：
在業務上規避這類問題
例如一個假設訂單場景，北京和上海兩個城市訂單量增長幾十倍，其余城市的數據量不變。這時候我們在進行聚合的時候，北京和上海就會出現數據堆積，我們可以單獨數據北京和上海的數據。
Key的設計上
把熱key進行拆分，比如上個例子中的北京和上海，可以把北京和上海按照地區進行拆分聚合。
參數設置
Flink 1.9.0 SQL(Blink Planner) 性能優化中一項重要的改進就是升級了微批模型，即 MiniBatch。原理是緩存一定的數據后再觸發處理，以減少對State的訪問，從而提升吞吐和減少數據的輸出量。
十一、Flink任務延遲高，想解決這個問題，你會如何入手？
在Flink的后台任務管理中，我們可以看到Flink的哪個算子和task出現了反壓。最主要的手段是資源調優和算子調優。資源調優即是對作業中的Operator的並發數（parallelism）、CPU（core）、堆內存（heap_memory）等參數進行調優。作業參數調優包括：並行度的設置，State的設置，checkpoint的設置。
十二、Flink是如何處理反壓的？
1.5版本的TCP的反壓，是通過callback實現的，當socket發送數據去receive buffer后，receiver會反饋給send端，目前receiver端的buffer還有多少剩余空間，然后send會根據剩余空間，控制發送速率。
TCP這種方式的弊端：
1.因為TM中會有多個Task運行，所以單個Task的反壓會阻斷整個TM的socket，而其他的task也無法向下游發送數據，連checkpoint的barrier也無法發出。
2.反壓傳播路徑長，導致生效延遲比較大。
1.5版本后采用的credit這種反壓機制是在Flink層面上的：
ResultSubpartition會向InputGate發送這個要發送的量，InputGate返回當前空余量，包含LocalBufferPool的。如果這個時候發現backlog > credit，那么LocalBufferPool就會向NetWorkPool申請內存。
長此以往，當credit返回0的時候，表示沒有內存緩存了，那么ResultSubpartition接收到credit的時候，就不會繼續往netty寫數據了。這樣socket就不會堵塞了，然后生效延遲也降低了。同時ResultPartition也會不斷去探測InputGate是否有空余的空間。
十三、 Operator Chains（算子鏈）這個概念你了解嗎？
Flink會在生成JobGraph階段，將代碼中可以優化的算子優化成一個算子鏈（Operator Chains）以放到一個task（一個線程）中執行：以減少線程之間的切換和緩沖的開銷，提高整體的吞吐量和延遲。
十四、 Flink什么情況下才會把Operator chain在一起形成算子鏈？
兩個operator chain在一起的的條件：
上下游的並行度一致
下游節點的入度為1 （也就是說下游節點沒有來自其他節點的輸入）
上下游節點都在同一個 slot group 中（下面會解釋 slot group）
下游節點的 chain 策略為 ALWAYS（可以與上下游鏈接，map、flatmap、filter等默認是ALWAYS）
上游節點的 chain 策略為 ALWAYS 或 HEAD（只能與下游鏈接，不能與上游鏈接，Source默認是HEAD）
兩個節點間數據分區方式是 forward（參考理解數據流的分區）
用戶沒有禁用 chain
十五、 說說Flink1.9的新特性？
支持hive讀寫，支持UDF
Flink SQL TopN和GroupBy等優化
Checkpoint跟savepoint針對實際業務場景做了優化
Flink state查詢
十六、消費kafka數據的時候，如何處理臟數據？
可以在處理前加一個fliter算子，將不符合規則的數據過濾出去。