在flink中如何進(jìn)行keyby窗口數(shù)據(jù)傾斜的優(yōu)化

在大數(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ù)，范圍不能太大，也不能太小，太大的話，分的組太多，增加checkpoint的壓力，太小的話，起不到打散的作用。在我的測(cè)試中，一天大概十幾億的數(shù)據(jù)量，5個(gè)并行度，隨機(jī)數(shù)的范圍在100范圍內(nèi)，就可以正常處理了。

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