在大數(shù)據(jù)處理領(lǐng)域，數(shù)據(jù)傾斜是一個非常常見的問題，今天我們就簡單講講在flink中如何處理流式數(shù)據(jù)傾斜問題。

1.數(shù)據(jù)傾斜的原理和影響

1.1 原理

數(shù)據(jù)傾斜就是數(shù)據(jù)的分布嚴(yán)重不均，造成一部分?jǐn)?shù)據(jù)很多，一部分?jǐn)?shù)據(jù)很少的局面。

數(shù)據(jù)分布理論上都是傾斜的，符合“二八原理”：例如80%的財富集中在20%的人手中、80%的用戶只使用20%的功能、20%的用戶貢獻了80%的訪問量。

數(shù)據(jù)傾斜的現(xiàn)象，如下圖所示。

1.2 影響

（1）單點問題

數(shù)據(jù)集中在某些分區(qū)上（Subtask），導(dǎo)致數(shù)據(jù)嚴(yán)重不平衡。

（2）GC 頻繁

過多的數(shù)據(jù)集中在某些 JVM（TaskManager），使得JVM 的內(nèi)存資源短缺，導(dǎo)致頻繁 GC。

（3）吞吐下降、延遲增大

數(shù)據(jù)單點和頻繁 GC 導(dǎo)致吞吐下降、延遲增大。

（4）系統(tǒng)崩潰

嚴(yán)重情況下，過長的 GC 導(dǎo)致 TaskManager 失聯(lián)，系統(tǒng)崩潰。

2.Flink 如何定位數(shù)據(jù)傾斜？

步驟1：定位反壓

定位反壓有2種方式：Flink Web UI 自帶的反壓監(jiān)控（直接方式）、Flink Task Metrics（間接方式）。通過監(jiān)控反壓的信息，可以獲取到數(shù)據(jù)處理瓶頸的 Subtask。

參考：【Flink 精選】如何分析及處理反壓?

步驟2：確定數(shù)據(jù)傾斜

Flink Web UI 自帶Subtask 接收和發(fā)送的數(shù)據(jù)量。當(dāng) Subtasks 之間處理的數(shù)據(jù)量有較大的差距，則該 Subtask 出現(xiàn)數(shù)據(jù)傾斜。如下圖所示，紅框內(nèi)的 Subtask 出現(xiàn)數(shù)據(jù)熱點。

3.Flink 如何處理常見數(shù)據(jù)傾斜？

場景一：數(shù)據(jù)源 source 消費不均勻

解決思路：通過調(diào)整并發(fā)度，解決數(shù)據(jù)源消費不均勻或者數(shù)據(jù)源反壓的情況。

例如kafka數(shù)據(jù)源，可以調(diào)整 KafkaSource 的并發(fā)度解決消費不均勻。

調(diào)整并發(fā)度的原則：KafkaSource 并發(fā)度與 kafka 分區(qū)數(shù)是一樣的，或者 kafka 分區(qū)數(shù)是KafkaSource 并發(fā)度的整數(shù)倍。

場景二：key 分布不均勻的無統(tǒng)計場景

說明：key 分布不均勻的無統(tǒng)計場景，例如上游數(shù)據(jù)分布不均勻，使用keyBy來打散數(shù)據(jù)。

解決思路： 通過添加隨機前綴，打散 key 的分布，使得數(shù)據(jù)不會集中在幾個 Subtask。

具體措施：
① 在原來分區(qū) key/uid 的基礎(chǔ)上，加上隨機的前綴或者后綴。
②使用數(shù)據(jù)到達的順序seq，作為分區(qū)的key。

場景三：key 分布不均勻的統(tǒng)計場景

解決思路：聚合統(tǒng)計前，先進行預(yù)聚合，例如兩階段聚合（加鹽局部聚合+去鹽全局聚合）。

兩階段聚合的具體措施：
①預(yù)聚合：加鹽局部聚合，在原來的 key 上加隨機的前綴或者后綴。
②聚合：去鹽全局聚合，刪除預(yù)聚合添加的前綴或者后綴，然后進行聚合統(tǒng)計。

4.一個SQL演示案例

我們先來看一個可能產(chǎn)生數(shù)據(jù)傾斜的sql

select TUMBLE_END(proc_time, INTERVAL '1' MINUTE) as winEnd,plat,count(*) as pv  from source_kafka_table 
group by TUMBLE(proc_time, INTERVAL '1' MINUTE) ,plat

在這個sql里，我們統(tǒng)計一個網(wǎng)站各個端的每分鐘的pv，從kafka消費過來的數(shù)據(jù)首先會按照端進行分組，然后執(zhí)行聚合函數(shù)count來進行pv的計算。

如果某一個端產(chǎn)生的數(shù)據(jù)特別大，比如我們的微信小程序端產(chǎn)生數(shù)據(jù)遠(yuǎn)遠(yuǎn)大于其他app端的數(shù)據(jù)，那么把這些數(shù)據(jù)分組到某一個算子之后，由于這個算子的處理速度跟不上，就會產(chǎn)生數(shù)據(jù)傾斜。

查看flink的ui，會看到如下的場景。

對于這種簡單的數(shù)據(jù)傾斜，我們可以通過對分組的key加上隨機數(shù)，再次打散，分別計算打散后不同的分組的pv數(shù)，然后在最外層再包一層，把打散的數(shù)據(jù)再次聚合，這樣就解決了數(shù)據(jù)傾斜的問題。

優(yōu)化后的sql如下：

select winEnd,split_index(plat1,'_',0) as plat2,sum(pv) from (

  select TUMBLE_END(proc_time, INTERVAL '1' MINUTE) as winEnd,plat1,count(*) as pv from (

    -- 最內(nèi)層，將分組的key，也就是plat加上一個隨機數(shù)打散
    select plat || '_' || cast(cast(RAND()*100 as int) as string) as plat1 ,proc_time from source_kafka_table 

) group by TUMBLE(proc_time, INTERVAL '1' MINUTE), plat1

) group by winEnd,split_index(plat1,'_',0)

在這個sql的最內(nèi)層，將分組的key，也就是plat加上一個隨機數(shù)打散，然后求打散后的各個分組（也就是sql中的plat1）的pv值，然后最外層，將各個打散的pv求和。

注意：最內(nèi)層的sql，給分組的key添加的隨機數(shù)，范圍不能太大，也不能太小，太大的話，分的組太多，增加checkpoint的壓力，太小的話，起不到打散的作用。

在我的測試中，一天大概十幾億的數(shù)據(jù)量，5個并行度，隨機數(shù)的范圍在100范圍內(nèi)，就可以正常處理了。

修改后我們看到各個子任務(wù)的數(shù)據(jù)基本均勻了。

https://www.cnblogs.com/qiu-hua/p/14056747.html

https://www.cnblogs.com/qiu-hua/p/14056780.html

歡迎關(guān)注微信公眾號：大數(shù)據(jù)從業(yè)者

總結(jié)

以上是生活随笔為你收集整理的Flink数据倾斜概述与优化的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。