Flink的序列化怎么做

本篇內(nèi)容主要講解“Flink的序列化怎么做”，感興趣的朋友不妨來(lái)看看。本文介紹的方法操作簡(jiǎn)單快捷，實(shí)用性強(qiáng)。下面就讓小編來(lái)帶大家學(xué)習(xí)“Flink的序列化怎么做”吧!

一、Flink是如何支持批流一體的？ 

本道面試題考察的其實(shí)就是一句話：Flink的開(kāi)發(fā)者認(rèn)為批處理是流處理的一種特殊情況。批處理是有限的流處理。Flink 使用一個(gè)引擎支持了DataSet API 和 DataStream API。

二、Flink是如何做到高效的數(shù)據(jù)交換的？ 

在一個(gè)Flink Job中，數(shù)據(jù)需要在不同的task中進(jìn)行交換，整個(gè)數(shù)據(jù)交換是有 TaskManager 負(fù)責(zé)的，TaskManager 的網(wǎng)絡(luò)組件首先從緩沖buffer中收集records，然后再發(fā)送。Records 并不是一個(gè)一個(gè)被發(fā)送的，二是積累一個(gè)批次再發(fā)送，batch 技術(shù)可以更加高效的利用網(wǎng)絡(luò)資源。

三、Flink是如何做容錯(cuò)的？

Flink 實(shí)現(xiàn)容錯(cuò)主要靠強(qiáng)大的CheckPoint機(jī)制和State機(jī)制。Checkpoint 負(fù)責(zé)定時(shí)制作分布式快照、對(duì)程序中的狀態(tài)進(jìn)行備份；State 用來(lái)存儲(chǔ)計(jì)算過(guò)程中的中間狀態(tài)。

四、Flink 分布式快照的原理是什么？

Flink的分布式快照是根據(jù)Chandy-Lamport算法量身定做的。簡(jiǎn)單來(lái)說(shuō)就是持續(xù)創(chuàng)建分布式數(shù)據(jù)流及其狀態(tài)的一致快照。

核心思想是在 input source 端插入 barrier，控制 barrier 的同步來(lái)實(shí)現(xiàn) snapshot 的備份和 exactly-once 語(yǔ)義。

五、Flink 是如何保證Exactly-once語(yǔ)義的？

Flink通過(guò)實(shí)現(xiàn)兩階段提交和狀態(tài)保存來(lái)實(shí)現(xiàn)端到端的一致性語(yǔ)義。
分為以下幾個(gè)步驟：

• 開(kāi)始事務(wù)（beginTransaction）創(chuàng)建一個(gè)臨時(shí)文件夾，來(lái)寫(xiě)把數(shù)據(jù)寫(xiě)入到這個(gè)文件夾里面

• 預(yù)提交（preCommit）將內(nèi)存中緩存的數(shù)據(jù)寫(xiě)入文件并關(guān)閉

• 正式提交（commit）將之前寫(xiě)完的臨時(shí)文件放入目標(biāo)目錄下。這代表著最終的數(shù)據(jù)會(huì)有一些延遲

• 丟棄（abort）丟棄臨時(shí)文件

若失敗發(fā)生在預(yù)提交成功后，正式提交前?？梢愿鶕?jù)狀態(tài)來(lái)提交預(yù)提交的數(shù)據(jù)，也可刪除預(yù)提交的數(shù)據(jù)。

六、Flink 的 kafka 連接器有什么特別的地方？

Flink源碼中有一個(gè)獨(dú)立的connector模塊，所有的其他connector都依賴于此模塊，F(xiàn)link 在1.9版本發(fā)布的全新kafka連接器，摒棄了之前連接不同版本的kafka集群需要依賴不同版本的connector這種做法，只需要依賴一個(gè)connector即可。

七、說(shuō)說(shuō) Flink的內(nèi)存管理是如何做的? 

Flink 并不是將大量對(duì)象存在堆上，而是將對(duì)象都序列化到一個(gè)預(yù)分配的內(nèi)存塊上。此外，F(xiàn)link大量的使用了堆外內(nèi)存。如果需要處理的數(shù)據(jù)超出了內(nèi)存限制，則會(huì)將部分?jǐn)?shù)據(jù)存儲(chǔ)到硬盤(pán)上。
Flink 為了直接操作二進(jìn)制數(shù)據(jù)實(shí)現(xiàn)了自己的序列化框架。

理論上Flink的內(nèi)存管理分為三部分：

• Network Buffers：這個(gè)是在TaskManager啟動(dòng)的時(shí)候分配的，這是一組用于緩存網(wǎng)絡(luò)數(shù)據(jù)的內(nèi)存，每個(gè)塊是32K，默認(rèn)分配2048個(gè)，可以通過(guò)“taskmanager.network.numberOfBuffers”修改

• Memory Manage pool：大量的Memory Segment塊，用于運(yùn)行時(shí)的算法（Sort/Join/Shuffle等），這部分啟動(dòng)的時(shí)候就會(huì)分配。下面這段代碼，根據(jù)配置文件中的各種參數(shù)來(lái)計(jì)算內(nèi)存的分配方法。（heap or off-heap，這個(gè)放到下節(jié)談），內(nèi)存的分配支持預(yù)分配和lazy load，默認(rèn)懶加載的方式。

• User Code，這部分是除了Memory Manager之外的內(nèi)存用于User code和TaskManager本身的數(shù)據(jù)結(jié)構(gòu)。

八、說(shuō)說(shuō) Flink的序列化如何做的? 

Java本身自帶的序列化和反序列化的功能，但是輔助信息占用空間比較大，在序列化對(duì)象時(shí)記錄了過(guò)多的類(lèi)信息。

Apache Flink摒棄了Java原生的序列化方法，以獨(dú)特的方式處理數(shù)據(jù)類(lèi)型和序列化，包含自己的類(lèi)型描述符，泛型類(lèi)型提取和類(lèi)型序列化框架。

TypeInformation 是所有類(lèi)型描述符的基類(lèi)。它揭示了該類(lèi)型的一些基本屬性，并且可以生成序列化器。TypeInformation 支持以下幾種類(lèi)型：

• BasicTypeInfo: 任意Java 基本類(lèi)型或 String 類(lèi)型

• BasicArrayTypeInfo: 任意Java基本類(lèi)型數(shù)組或 String 數(shù)組

• WritableTypeInfo: 任意 Hadoop Writable 接口的實(shí)現(xiàn)類(lèi)

• TupleTypeInfo: 任意的 Flink Tuple 類(lèi)型(支持Tuple1 to Tuple25)。Flink tuples 是固定長(zhǎng)度固定類(lèi)型的Java Tuple實(shí)現(xiàn)

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

• PojoTypeInfo: 任意的 POJO (Java or Scala)，例如，Java對(duì)象的所有成員變量，要么是 public 修飾符定義，要么有 getter/setter 方法

• GenericTypeInfo: 任意無(wú)法匹配之前幾種類(lèi)型的類(lèi)

針對(duì)前六種類(lèi)型數(shù)據(jù)集，F(xiàn)link皆可以自動(dòng)生成對(duì)應(yīng)的TypeSerializer，能非常高效地對(duì)數(shù)據(jù)集進(jìn)行序列化和反序列化。

九、 Flink中的Window出現(xiàn)了數(shù)據(jù)傾斜，你有什么解決辦法？ 

window產(chǎn)生數(shù)據(jù)傾斜指的是數(shù)據(jù)在不同的窗口內(nèi)堆積的數(shù)據(jù)量相差過(guò)多。本質(zhì)上產(chǎn)生這種情況的原因是數(shù)據(jù)源頭發(fā)送的數(shù)據(jù)量速度不同導(dǎo)致的。出現(xiàn)這種情況一般通過(guò)兩種方式來(lái)解決：

• 在數(shù)據(jù)進(jìn)入窗口前做預(yù)聚合

• 重新設(shè)計(jì)窗口聚合的key

十、 Flink中在使用聚合函數(shù) GroupBy、Distinct、KeyBy 等函數(shù)時(shí)出現(xiàn)數(shù)據(jù)熱點(diǎn)該如何解決？

數(shù)據(jù)傾斜和數(shù)據(jù)熱點(diǎn)是所有大數(shù)據(jù)框架繞不過(guò)去的問(wèn)題。處理這類(lèi)問(wèn)題主要從3個(gè)方面入手：

• 在業(yè)務(wù)上規(guī)避這類(lèi)問(wèn)題

例如一個(gè)假設(shè)訂單場(chǎng)景，北京和上海兩個(gè)城市訂單量增長(zhǎng)幾十倍，其余城市的數(shù)據(jù)量不變。這時(shí)候我們?cè)谶M(jìn)行聚合的時(shí)候，北京和上海就會(huì)出現(xiàn)數(shù)據(jù)堆積，我們可以單獨(dú)數(shù)據(jù)北京和上海的數(shù)據(jù)。

• Key的設(shè)計(jì)上

把熱key進(jìn)行拆分，比如上個(gè)例子中的北京和上海，可以把北京和上海按照地區(qū)進(jìn)行拆分聚合。

• 參數(shù)設(shè)置

Flink 1.9.0 SQL(Blink Planner) 性能優(yōu)化中一項(xiàng)重要的改進(jìn)就是升級(jí)了微批模型，即 MiniBatch。原理是緩存一定的數(shù)據(jù)后再觸發(fā)處理，以減少對(duì)State的訪問(wèn)，從而提升吞吐和減少數(shù)據(jù)的輸出量。

十一、Flink任務(wù)延遲高，想解決這個(gè)問(wèn)題，你會(huì)如何入手？ 

在Flink的后臺(tái)任務(wù)管理中，我們可以看到Flink的哪個(gè)算子和task出現(xiàn)了反壓。最主要的手段是資源調(diào)優(yōu)和算子調(diào)優(yōu)。資源調(diào)優(yōu)即是對(duì)作業(yè)中的Operator的并發(fā)數(shù)（parallelism）、CPU（core）、堆內(nèi)存（heap_memory）等參數(shù)進(jìn)行調(diào)優(yōu)。作業(yè)參數(shù)調(diào)優(yōu)包括：并行度的設(shè)置，State的設(shè)置，checkpoint的設(shè)置。

十二、Flink是如何處理反壓的？

Flink 內(nèi)部是基于 producer-consumer 模型來(lái)進(jìn)行消息傳遞的，F(xiàn)link的反壓設(shè)計(jì)也是基于這個(gè)模型。Flink 使用了高效有界的分布式阻塞隊(duì)列，就像 Java 通用的阻塞隊(duì)列（BlockingQueue）一樣。下游消費(fèi)者消費(fèi)變慢，上游就會(huì)受到阻塞。

十三、Flink的反壓和Strom有哪些不同？

Storm 是通過(guò)監(jiān)控 Bolt 中的接收隊(duì)列負(fù)載情況，如果超過(guò)高水位值就會(huì)將反壓信息寫(xiě)到 Zookeeper ，Zookeeper 上的 watch 會(huì)通知該拓?fù)涞乃?Worker 都進(jìn)入反壓狀態(tài)，最后 Spout 停止發(fā)送 tuple。

Flink中的反壓使用了高效有界的分布式阻塞隊(duì)列，下游消費(fèi)變慢會(huì)導(dǎo)致發(fā)送端阻塞。

二者最大的區(qū)別是Flink是逐級(jí)反壓，而Storm是直接從源頭降速。

十四、 Operator Chains（算子鏈）這個(gè)概念你了解嗎？

為了更高效地分布式執(zhí)行，F(xiàn)link會(huì)盡可能地將operator的subtask鏈接（chain）在一起形成task。每個(gè)task在一個(gè)線程中執(zhí)行。將operators鏈接成task是非常有效的優(yōu)化：它能減少線程之間的切換，減少消息的序列化/反序列化，減少數(shù)據(jù)在緩沖區(qū)的交換，減少了延遲的同時(shí)提高整體的吞吐量。這就是我們所說(shuō)的算子鏈。

十五、 Flink什么情況下才會(huì)把Operator chain在一起形成算子鏈？

兩個(gè)operator chain在一起的的條件：

• 上下游的并行度一致

• 下游節(jié)點(diǎn)的入度為1 （也就是說(shuō)下游節(jié)點(diǎn)沒(méi)有來(lái)自其他節(jié)點(diǎn)的輸入）

• 上下游節(jié)點(diǎn)都在同一個(gè) slot group 中（下面會(huì)解釋 slot group）

• 下游節(jié)點(diǎn)的 chain 策略為 ALWAYS（可以與上下游鏈接，map、flatmap、filter等默認(rèn)是ALWAYS）

• 上游節(jié)點(diǎn)的 chain 策略為 ALWAYS 或 HEAD（只能與下游鏈接，不能與上游鏈接，Source默認(rèn)是HEAD）

• 兩個(gè)節(jié)點(diǎn)間數(shù)據(jù)分區(qū)方式是 forward（參考理解數(shù)據(jù)流的分區(qū)）

• 用戶沒(méi)有禁用 chain

十六、 說(shuō)說(shuō)Flink1.9的新特性？

• 支持hive讀寫(xiě)，支持UDF

• Flink SQL TopN和GroupBy等優(yōu)化

• Checkpoint跟savepoint針對(duì)實(shí)際業(yè)務(wù)場(chǎng)景做了優(yōu)化

• Flink state查詢

十七、消費(fèi)kafka數(shù)據(jù)的時(shí)候，如何處理臟數(shù)據(jù)？

可以在處理前加一個(gè)fliter算子，將不符合規(guī)則的數(shù)據(jù)過(guò)濾出去。

到此，相信大家對(duì)“Flink的序列化怎么做”有了更深的了解，不妨來(lái)實(shí)際操作一番吧！這里是億速云網(wǎng)站，更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢，關(guān)注我們，繼續(xù)學(xué)習(xí)！