在大數(shù)據(jù)處理領(lǐng)域枝嘶,數(shù)據(jù)傾斜是一個(gè)非常常見(jiàn)的問(wèn)題,今天我們就簡(jiǎn)單講講在flink中如何處理流式數(shù)據(jù)傾斜問(wèn)題。
我們先來(lái)看一個(gè)可能產(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
在這個(gè)sql里帮碰,我們統(tǒng)計(jì)一個(gè)網(wǎng)站各個(gè)端的每分鐘的pv代兵,從kafka消費(fèi)過(guò)來(lái)的數(shù)據(jù)首先會(huì)按照端進(jìn)行分組尼酿,然后執(zhí)行聚合函數(shù)count來(lái)進(jìn)行pv的計(jì)算。如果某一個(gè)端產(chǎn)生的數(shù)據(jù)特別大植影,比如我們的微信小程序端產(chǎn)生數(shù)據(jù)遠(yuǎn)遠(yuǎn)大于其他app端的數(shù)據(jù)裳擎,那么把這些數(shù)據(jù)分組到某一個(gè)算子之后,由于這個(gè)算子的處理速度跟不上思币,就會(huì)產(chǎn)生數(shù)據(jù)傾斜鹿响。
查看flink的ui,會(huì)看到如下的場(chǎng)景支救。
對(duì)于這種簡(jiǎn)單的數(shù)據(jù)傾斜,我們可以通過(guò)對(duì)分組的key加上隨機(jī)數(shù)各墨,再次打散,分別計(jì)算打散后不同的分組的pv數(shù)贬堵,然后在最外層再包一層,把打散的數(shù)據(jù)再次聚合黎做,這樣就解決了數(shù)據(jù)傾斜的問(wèn)題叉跛。
優(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加上一個(gè)隨機(jī)數(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)
在這個(gè)sql的最內(nèi)層鸣峭,將分組的key酥艳,也就是plat加上一個(gè)隨機(jī)數(shù)打散摊溶,然后求打散后的各個(gè)分組(也就是sql中的plat1)的pv值,然后最外層充石,將各個(gè)打散的pv求和莫换。
注意:最內(nèi)層的sql,給分組的key添加的隨機(jī)數(shù)骤铃,范圍不能太大,也不能太小惰爬,太大的話(huà),分的組太多撕瞧,增加checkpoint的壓力,太小的話(huà)风范,起不到打散的作用。在我的測(cè)試中硼婿,一天大概十幾億的數(shù)據(jù)量,5個(gè)并行度寇漫,隨機(jī)數(shù)的范圍在100范圍內(nèi),就可以正常處理了州胳。
修改后我們看到各個(gè)子任務(wù)的數(shù)據(jù)基本均勻了。
更多干貨信息遍膜,歡迎關(guān)注我的公眾號(hào)【大數(shù)據(jù)技術(shù)與應(yīng)用實(shí)戰(zhàn)】