問題描述
近期業務反饋， 開啟了 mini-batch 之后， 出現了資料不準的情況， 關掉了 mini-batch 之后， 就正常了， 因此業務方懷疑，是不是 Flink 的 mini-batch 存在 bug ？
問題排查
初步分析
- mini-batch 已經在內部大規模使用， 目前沒有發現一例和開啟 mini-batch 有關， 同時 mini-batch 本質只是將資料進行攢批然后計算， 并沒有修改核心的運算邏輯.
- 開關 mini-batch 的關鍵時資料的批量計算， 是否在批量計算使得原本存在 bug 的代碼暴露問題
- 業務在 Flink SQL 使用了多個雙流 join 和 group window，如果不注意使用，很可能導致亂序，最終的錯誤結果是某條資料沒有被正常更新， 和亂序的情況比較類似.
綜上考慮， 整體排查的方向還是排查 SQL 的業務邏輯是否存在亂序的 case， 開啟了 mini-batch 后是否加劇了這種亂序的產生
代碼邏輯梳理
flowchart LR join1(join1 \n item_day, item_key) --> join2 join2(join2 \n item_day, item_key) --> join3 join3(join3 \n item_day, item_key) --> group1 group1(group1 \n item_day, item_key) --> group2 group2(group2 \n item_day, item_key, key1, key2, key3) --> sink sink(sink \n pk: item_day, item_key)抽象之后的 DAG 如圖所示:
- join1, join2, join3, group1 都是基于 item_day 和 item_key 進行 hash 資料經過這些算子均按照 [item_day, item_key] 進行 hash
- group2 算子的 group key 為 [item_day, item_key, key1， key2， key3]，Flink 會基于這些欄位整體進行 hash
- Sink 算子的主鍵為 [item_day, item_key] ，資料流向 Sink 算子時會按照 [item_day, item_key] 進行 hash.
分析:
key1, key2, key3 時由前面的 join1 算子補充的維度欄位， 前面的 join 采用的是 left join， 因此可能會存在 item_day 和 item_key 相同的資料， 對應的 key1, key2, key3 并不相同， 經過 group2 會觸發具有相同 [item_day, item_key] 的資料，被 hash 到不同的并發，這種就出現了亂序問題
修復手段
最后的 group by [item_day, item_key, key1, key2, key3]， 核心還是為了聚合相同的 item_day和 item_key， key1, key2, key3 不屬于 value 型別資料， 也不參與聚合， 因此將修改 SQL 避免基于 key1, key2, key3 進行聚合即可， 這里采用 last_value 聚合函式取最后一條資料
-- 原始 SQL
SELECT item_day, item_key, key1, key2, key3, sum(value)
FROM XXX
GROUP BY item_day, item_key, key1, key2, key3
-- 修改為
SELECT item_day, item_key, last_value(key1), last_value(key2), last_value(key3), sum(value)
FROM XXX
GROUP BY item_day, item_key
經過修改之后，保證整個 Flink 處理鏈路中， 相同的主鍵對應的資料，無論經過多少次 hash， 都是在同一個并行處理，這種才能保證最終結果的正確性
結論
修改后， 業務的結果恢復正常， 因此 Mini-batch 并不是導致作業出現問題的核心原因， 核心原因還是亂序， 而開啟 mini-batch 會加劇這種亂序問題的觸發，
開啟 mini-batch 之后， 具有相同 key 的資料， 如果落到了同一個 batch， 這樣物理上的時間差就更短，因而更容易暴露問題，
轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/541165.html
標籤：其他