ToB企服应用市场:ToB评测及商务社交产业平台

标题: Flink mini-batch "引发" 的乱序问题 [打印本页]

作者: 来自云龙湖轮廓分明的月亮    时间: 2023-1-4 10:16
标题: Flink mini-batch "引发" 的乱序问题
问题描述

近期业务反馈, 开启了 mini-batch 之后, 出现了数据不准的情况, 关掉了 mini-batch 之后, 就正常了, 因此业务方怀疑,是不是 Flink 的 mini-batch 存在 bug ?
问题排查

初步分析

综上考虑, 整体排查的方向还是排查 SQL 的业务逻辑是否存在乱序的 case, 开启了 mini-batch 后是否加剧了这种乱序的产生
代码逻辑梳理

flowchart LRjoin1(join1 \n item_day, item_key) --> join2join2(join2 \n item_day, item_key) --> join3join3(join3 \n item_day, item_key) --> group1group1(group1 \n item_day, item_key) --> group2group2(group2 \n item_day, item_key, key1, key2, key3) --> sinksink(sink \n pk: item_day, item_key)抽象之后的 DAG 如图所示:
分析:
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 聚合函数取最后一条数据
  1. -- 原始 SQL
  2. SELECT item_day, item_key, key1, key2, key3, sum(value)
  3. FROM XXX
  4. GROUP BY item_day, item_key, key1, key2, key3
  5. -- 修改为
  6. SELECT item_day, item_key, last_value(key1), last_value(key2), last_value(key3), sum(value)
  7. FROM XXX
  8. GROUP BY item_day, item_key
复制代码
经过修改之后,保证整个 Flink 处理链路中, 相同的主键对应的数据,无论经过多少次 hash, 都是在同一个并行处理,这种才能保证最终结果的正确性
结论

修改后, 业务的结果恢复正常, 因此 Mini-batch 并不是导致作业出现问题的核心原因, 核心原因还是乱序, 而开启 mini-batch 会加剧这种乱序问题的触发。
开启 mini-batch 之后, 具有相同 key 的数据, 如果落到了同一个 batch, 这样物理上的时间差就更短,因而更容易暴露问题。

免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!




欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4