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

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

我們先來看一個可能產(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ù)基本均勻了。