數據傾斜原理

數據傾斜就是數據的分布嚴重不均，流入部分算子的數據明顯多余其他算子，造成這部分算子壓力過(guò)大。

影響

單點(diǎn)問(wèn)題

數據集中在某些分區上（Subtask），導致數據嚴重不平衡。

GC 頻繁

過(guò)多的數據集中在某些 JVM（TaskManager），使得JVM 的內存資源短缺，導致頻繁 GC。

吞吐下降、延遲增大

數據單點(diǎn)和頻繁 GC 導致吞吐下降、延遲增大。

系統崩潰

嚴重情況下，過(guò)長(cháng)的 GC 導致 TaskManager 失聯(lián)，系統崩潰。

Flink數據傾斜問(wèn)題定位

定位反壓

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

確定數據傾斜

Flink Web UI 自帶Subtask 接收和發(fā)送的數據量。當 Subtasks 之間處理的數據量有較大的差距，則該 Subtask 出現數據傾斜。

Flink 如何處理常見(jiàn)數據傾斜

數據源 source 消費不均勻

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

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

調整并發(fā)度的原則：KafkaSource 并發(fā)度與 kafka 分區數是一樣的，或者 kafka 分區數是KafkaSource 并發(fā)度的整數倍。

key 分布不均勻的無(wú)統計場(chǎng)景

說(shuō)明：key 分布不均勻的無(wú)統計場(chǎng)景，例如上游數據分布不均勻，使用keyBy來(lái)打散數據。

解決思路： 通過(guò)添加隨機前綴，打散 key 的分布，使得數據不會(huì )集中在幾個(gè) Subtask。

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

key 分布不均勻的統計場(chǎng)景

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

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

SQL 樣例

在下面SQL里面，我們統計一個(gè)網(wǎng)站各個(gè)端的每分鐘的pv，從kafka消費過(guò)來(lái)的數據首先會(huì )按照端進(jìn)行分組，然后執行聚合函
數count來(lái)進(jìn)行pv的計算。

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è)端產(chǎn)生的數據特別大，比如我們的微信小程序端產(chǎn)生數據遠遠大于其他app端的數據，那么把這些數據分組到某一
個(gè)算子之后，由于這個(gè)算子的處理速度跟不上，就會(huì )產(chǎn)生數據傾斜。

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 (
    -- 最內層，將分組的key，也就是plat加上一個(gè)隨機數打散
    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的最內層，將分組的key，也就是plat加上一個(gè)隨機數打散，然后求打散后的各個(gè)分組（也就是sql中的plat1）的
pv值，然后最外層，將各個(gè)打散的pv求和。

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