踩坑記 | Flink Sql Count 還有這種坑！
本文轉(zhuǎn)載自微信公眾號(hào)「大數(shù)據(jù)羊說(shuō)」，作者antigeneral了呀。轉(zhuǎn)載本文請(qǐng)聯(lián)系大數(shù)據(jù)羊說(shuō)公眾號(hào)。
1.序篇
通過(guò)本文你可了解到
- 踩坑場(chǎng)景篇-這個(gè)坑是啥樣的
- 問(wèn)題排查篇-坑的排查過(guò)程
- 問(wèn)題原理解析篇-導(dǎo)致問(wèn)題的機(jī)制是什么
- 避坑篇-如何避免這種問(wèn)題
- 展望篇-有什么機(jī)制可以根本避免這種情況
先說(shuō)下結(jié)論：在非窗口類 flink sql 任務(wù)中，會(huì)存在 retract 機(jī)制，即上游會(huì)向下游發(fā)送「撤回消息(做減法)」，**最新的結(jié)果消息(做加法)**兩條消息來(lái)計(jì)算結(jié)果，保證結(jié)果正確性。
而如果我們?cè)谏舷掠沃虚g使用了映射類 udf 改變了**撤回消息(做減法)「的一些字段值時(shí)，就可能會(huì)導(dǎo)致」撤回消息(做減法)**不能被正常處理，最終導(dǎo)致結(jié)果的錯(cuò)誤。
2.踩坑場(chǎng)景篇-這個(gè)坑是啥樣的
在介紹坑之前我們先介紹下我們的需求、實(shí)現(xiàn)方案的背景。
2.1.背景
在各類游戲中都會(huì)有一種場(chǎng)景，一個(gè)用戶可以從 A 等級(jí)升級(jí)到 B 等級(jí)，用戶可以不斷的升級(jí)，但是一個(gè)用戶同一時(shí)刻只會(huì)在同一個(gè)等級(jí)。需求指標(biāo)就是當(dāng)前分鐘各個(gè)等級(jí)的用戶數(shù)。
2.2.預(yù)期效果
2
2.3.解決思路
獲取到當(dāng)前所有用戶的最新等級(jí)
一個(gè)用戶同一時(shí)刻只會(huì)在一個(gè)等級(jí)，所以對(duì)每一個(gè)等級(jí)的用戶做 count 操作
2.4.解決方案
獲取到當(dāng)前所有用戶的最新等級(jí)：flink sql row_number() 就可以實(shí)現(xiàn)，按照數(shù)據(jù)的 rowtime 進(jìn)行逆序排序就可以獲取到用戶當(dāng)前最新的等級(jí)
對(duì)每一個(gè)等級(jí)的用戶做 count 操作：對(duì) row_number() 的后的明細(xì)結(jié)果進(jìn)行 count 操作
2.4.1.sql
具體實(shí)現(xiàn) sql 如下，非常簡(jiǎn)單：
- WITH detail_tmp AS (
- SELECT
- 等級(jí),
- id,
- `timestamp`
- FROM
- (
- SELECT
- 等級(jí),
- id,
- `timestamp`,
- -- row_number 獲取最新狀態(tài)
- row_number() over(
- PARTITION by id
- ORDER BY
- `timestamp` DESC
- ) AS rn
- FROM
- source_db.source_table
- )
- WHERE
- rn = 1
- )
- SELECT
- DIM.中文等級(jí) as 等級(jí),
- sum(part_uv) as uv
- FROM
- (
- SELECT
- 等級(jí),
- count(id) as part_uv
- FROM
- detail_tmp
- GROUP BY
- 等級(jí),
- mod(id, 1024)
- )
- -- 上游數(shù)據(jù)的等級(jí)名稱是數(shù)字，需求方要求給轉(zhuǎn)換成中文，所以這里加了一個(gè) udf 映射
- LEFT JOIN LATERAL TABLE(等級(jí)中文映射_UDF(等級(jí))) AS DIM(中文等級(jí)) ON TRUE
- GROUP BY
- DIM.中文等級(jí)
2.4.2.參數(shù)配置
使用 minibatch 參數(shù)方式控制數(shù)據(jù)輸出頻率。
- table.exec.mini-batch.enabled : true
- -- 設(shè)定 60s 的觸發(fā)間隔
- table.exec.mini-batch.allow-latency : 60s
- table.exec.mini-batch.size : 10000000000
任務(wù) plan。
1
2.5.問(wèn)題場(chǎng)景
這段 SQL 跑了 n 年都沒(méi)有問(wèn)題，但是有一天運(yùn)營(yíng)在配置【等級(jí)中文映射_UDF】時(shí)，不小心將一個(gè)等級(jí)的中文名給映射錯(cuò)了，雖然馬上恢復(fù)了，但是當(dāng)天的實(shí)時(shí)數(shù)據(jù)和離線數(shù)據(jù)對(duì)比后卻發(fā)現(xiàn)，實(shí)時(shí)產(chǎn)出的數(shù)值比離線大很多!!!而之前都是保持一致的。
3.問(wèn)題排查篇-坑的排查過(guò)程
首先我們想一下，這個(gè)指標(biāo)是算 uv 的，運(yùn)營(yíng)將等級(jí)中文名配置錯(cuò)了，也應(yīng)該是把原有等級(jí)的最終結(jié)果算少啊，怎么會(huì)算多呢???
然后我們將場(chǎng)景復(fù)現(xiàn)了下，來(lái)看看代碼：
任務(wù)代碼，大家可以直接 copy 到本地運(yùn)行：
- public class Test {
- public static void main(String[] args) throws Exception {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setParallelism(1);
- EnvironmentSettings settings = EnvironmentSettings
- .newInstance()
- .useBlinkPlanner()
- .inStreamingMode().build();
- StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);
- // 模擬輸入
- DataStream<Tuple3<String, Long, Long>> tuple3DataStream =
- env.fromCollection(Arrays.asList(
- Tuple3.of("2", 1L, 1627218000000L),
- Tuple3.of("2", 101L, 1627218000000L + 6000L),
- Tuple3.of("2", 201L, 1627218000000L + 7000L),
- Tuple3.of("2", 301L, 1627218000000L + 7000L)));
- // 分桶取模 udf
- tEnv.registerFunction("mod", new Mod_UDF());
- // 中文映射 udf
- tEnv.registerFunction("status_mapper", new StatusMapper_UDF());
- tEnv.createTemporaryView("source_db.source_table", tuple3DataStream,
- "status, id, timestamp");
- String sql = "WITH detail_tmp AS (\n"
- + " SELECT\n"
- + " status,\n"
- + " id,\n"
- + " `timestamp`\n"
- + " FROM\n"
- + " (\n"
- + " SELECT\n"
- + " status,\n"
- + " id,\n"
- + " `timestamp`,\n"
- + " row_number() over(\n"
- + " PARTITION by id\n"
- + " ORDER BY\n"
- + " `timestamp` DESC\n"
- + " ) AS rn\n"
- + " FROM source_db.source_table"
- + " )\n"
- + " WHERE\n"
- + " rn = 1\n"
- + ")\n"
- + "SELECT\n"
- + " DIM.status_new as status,\n"
- + " sum(part_uv) as uv\n"
- + "FROM\n"
- + " (\n"
- + " SELECT\n"
- + " status,\n"
- + " count(distinct id) as part_uv\n"
- + " FROM\n"
- + " detail_tmp\n"
- + " GROUP BY\n"
- + " status,\n"
- + " mod(id, 100)\n"
- + " )\n"
- + "LEFT JOIN LATERAL TABLE(status_mapper(status)) AS DIM(status_new) ON TRUE\n"
- + "GROUP BY\n"
- + " DIM.status_new";
- Table result = tEnv.sqlQuery(sql);
- tEnv.toRetractStream(result, Row.class).print();
- env.execute();
- }
- }
UDF 代碼：
- public class StatusMapper_UDF extends TableFunction<String> {
- public void eval(String status) {
- if (status.equals("1")) {
- collector.collect("等級(jí)1");
- } else if (status.equals("2")) {
- collector.collect("等級(jí)2");
- } else if (status.equals("3")) {
- collector.collect("等級(jí)3");
- }
- }
- }
在正確情況(模擬 UDF 沒(méi)有任何變動(dòng)的情況下)的輸出結(jié)果：
- (true,等級(jí)2,1)
- (false,等級(jí)2,1)
- (true,等級(jí)2,2)
- (false,等級(jí)2,2)
- (true,等級(jí)2,3)
- (false,等級(jí)2,3)
- (true,等級(jí)2,4)
最終等級(jí)2 的 uv 數(shù)為 4，結(jié)果復(fù)合預(yù)期?。
模擬下用戶修改了 udf 配置之后，UDF 代碼如下：
- public class StatusMapper_UDF extends TableFunction<String> {
- private int i = 0;
- public void eval(String status) {
- if (i == 5) {
- collect("等級(jí)4");
- } else {
- if ("1".equals(status)) {
- collector.collect("等級(jí)1");
- } else if ("2".equals(status)) {
- collector.collect("等級(jí)2");
- } else if ("3".equals(status)) {
- collector.collect("等級(jí)3");
- }
- }
- i++;
- }
- }
得到的結(jié)果如下：
- (true,等級(jí)2,1)
- (false,等級(jí)2,1)
- (true,等級(jí)2,2)
- (false,等級(jí)2,2)
- (true,等級(jí)2,3)
- (false,等級(jí)2,3)
- (true,等級(jí)2,7)
最終等級(jí)2 的 uv 數(shù)為 7，很明顯這是錯(cuò)誤結(jié)果?。
因此可以確定是由于這個(gè) UDF 的處理邏輯變換而導(dǎo)致的結(jié)果出現(xiàn)錯(cuò)誤。
下文就讓我們來(lái)分析下其中緣由。
問(wèn)題原理解析篇-導(dǎo)致問(wèn)題的機(jī)制是什么
我們首先來(lái)分析下上述 SQL，可以發(fā)現(xiàn)整個(gè) flink sql 任務(wù)是使用了 unbounded + minibatch 實(shí)現(xiàn)的，在 minibatch 觸發(fā)條件觸發(fā)時(shí)，上游算子會(huì)將之前的結(jié)果撤回，然后將最新的結(jié)果發(fā)出。
這個(gè)任務(wù)的 execution plan 如圖所示。
7
可以從算子圖中的一些計(jì)算邏輯可以看到，整個(gè)任務(wù)都是基于 retract 機(jī)制運(yùn)行(count_retract、sum_retract 等)。
而涉及到 udf 的核心邏輯是在 Operator(ID = 7)，和 Operator(ID = 12) 之間。當(dāng) Operator(ID = 7) GroupAggregate 結(jié)果發(fā)生改變之后，會(huì)發(fā)一條「撤回消息(做減法)」，一條**最新的結(jié)果消息(做加法)**到 Operator(ID = 12) GroupAggregate。
5
Notes：簡(jiǎn)單解釋下上面說(shuō)的「撤回消息(做減法)」，「最新的結(jié)果消息(做加法)」。舉個(gè)算 count 的例子：當(dāng)整個(gè)任務(wù)的第一條數(shù)據(jù)來(lái)之后，之前沒(méi)有數(shù)據(jù)，所以不用撤回，結(jié)果就是 0(沒(méi)有數(shù)據(jù)) + 1(第一條數(shù)據(jù)) = 1(結(jié)果)，當(dāng)?shù)诙l結(jié)果來(lái)之后，就要將上次發(fā)的 1 消息(可以理解為是整個(gè)任務(wù)的一個(gè)中間結(jié)果)撤回，將最新的結(jié)果 2 發(fā)下去。那么計(jì)算方法就是 1(上次的結(jié)果) - 1(撤回) + 2(當(dāng)前最新的結(jié)果消息)= 2(結(jié)果)。
通過(guò)算子圖可以發(fā)現(xiàn)，【中文名稱映射】UDF 是處于兩個(gè) GroupAggregate 之間的。也就是說(shuō) Operator(ID = 7) GroupAggregate 發(fā)出的「撤回消息(做減法)」，**最新的結(jié)果消息(做加法)「都會(huì)執(zhí)行這個(gè) UDF，那么就有可能」撤回消息(做減法)「中的某個(gè)作為下游 GroupAggregate 算子 key 的字段會(huì)被更改成其他值，那么這條消息就不會(huì)發(fā)到原來(lái)下游 GroupAggregate 算子的原始 key 中，那么原來(lái)的 key 的歷史結(jié)果就撤回不了了。。。但是」最新的結(jié)果消息(做加法)**的字段沒(méi)有被更改時(shí)，那么這個(gè)消息依然被發(fā)到了下游 GroupAggregate 算子，這就會(huì)導(dǎo)致沒(méi)做減法，卻做了加法，就會(huì)導(dǎo)致結(jié)果增加，如下圖所示。
從這個(gè)角度出發(fā)，我們來(lái)分析下上面的 case，從內(nèi)層發(fā)給外層的消息一條一條來(lái)分析。
內(nèi)層消息怎么來(lái)看呢?其實(shí)就是將上面的 SQL 中的 left join 刪除，重新跑一遍就可以得到結(jié)果，結(jié)果如下：
- (true,等級(jí)2,1)
- (false,等級(jí)2,1)
- (true,等級(jí)2,2)
- (false,等級(jí)2,2)
- (true,等級(jí)2,3)
- (false,等級(jí)4,3)
- (true,等級(jí)2,4)
來(lái)分析下內(nèi)層消息發(fā)出之后對(duì)應(yīng)到外層消息的操作：
| 內(nèi)層 | 外層 |
|---|---|
| (true,等級(jí)2,1) | (true,等級(jí)2,1) |
| (false,等級(jí)2,1) | (false,等級(jí)2,1) |
| (true,等級(jí)2,2) | (true,等級(jí)2,2) |
| (false,等級(jí)2,2) | (false,等級(jí)2,2) |
| (true,等級(jí)2,3) | (true,等級(jí)2,3) |
前五條消息不會(huì)導(dǎo)致錯(cuò)誤，不用詳細(xì)說(shuō)明。
| 內(nèi)層 | 外層 |
|---|---|
| (true,等級(jí)2,1) | (true,等級(jí)2,1) |
| (false,等級(jí)2,1) | (false,等級(jí)2,1) |
| (true,等級(jí)2,2) | (true,等級(jí)2,2) |
| (false,等級(jí)2,2) | (false,等級(jí)2,2) |
| (true,等級(jí)2,3) | (true,等級(jí)2,3) |
| (false,等級(jí)4,3) |
第六條消息發(fā)出之后，經(jīng)過(guò) udf 的處理之后，中文名被映射成了【等級(jí)4】，而其再通過(guò) hash partition 策略向下發(fā)送消息時(shí)，就不能將這條撤回消息發(fā)到原本 key 為【等級(jí)2】的算子中了，這條撤回消息也無(wú)法被處理了。
| 內(nèi)層 | 外層 |
|---|---|
| (true,等級(jí)2,1) | (true,等級(jí)2,1) |
| (false,等級(jí)2,1) | (false,等級(jí)2,1) |
| (true,等級(jí)2,2) | (true,等級(jí)2,2) |
| (false,等級(jí)2,2) | (false,等級(jí)2,2) |
| (true,等級(jí)2,3) | (true,等級(jí)2,3) |
| (false,等級(jí)4,3) | |
| (true,等級(jí)2,4) | (false,等級(jí)2,3) (true,等級(jí)2,7) |
第七條消息 (true,等級(jí)2,4) 發(fā)出后，外層 GroupAggregate 算子首先會(huì)將上次發(fā)出的記過(guò)撤回，即(false,等級(jí)2,3)，然后將(true,等級(jí)2,4)累加到當(dāng)前的記過(guò)上，即 3(上次結(jié)果)+ 4(這次最新的結(jié)果)= 7(結(jié)果)。就導(dǎo)致了上述的錯(cuò)誤結(jié)果。
定位到問(wèn)題原因之后，我們來(lái)看看怎么避免上述錯(cuò)誤。
6.避坑篇-如何避免這種問(wèn)題
6.1.從源頭避免
udf 這種映射維度的 udf 盡量在上線前就固定下來(lái)，避免后續(xù)更改造成的數(shù)據(jù)錯(cuò)誤。
6.2.替換為 ScalarFunction 進(jìn)行映射
- WITH detail_tmp AS (
- SELECT
- status,
- id,
- `timestamp`
- FROM
- (
- SELECT
- status,
- id,
- `timestamp`,
- row_number() over(
- PARTITION by id
- ORDER BY
- `timestamp` DESC
- ) AS rn
- FROM
- (
- SELECT
- status,
- id,
- `timestamp`
- FROM
- source_db.source_table
- ) t1
- ) t2
- WHERE
- rn = 1
- )
- SELECT
- -- 在此處進(jìn)行中文名稱映射
- 等級(jí)中文映射_UDF(status) as status,
- sum(part_uv) as uv
- FROM
- (
- SELECT
- status,
- count(distinct id) as part_uv
- FROM
- detail_tmp
- GROUP BY
- status,
- mod(id, 100)
- )
- GROUP BY
- status
還是剛剛的邏輯，剛剛的配方，我們先來(lái)看一下結(jié)果。
- public class StatusMapper_UDF extends ScalarFunction {
- private int i = 0;
- public String eval(String status) {
- if (i == 5) {
- i++;
- return "等級(jí)4";
- } else {
- i++;
- if ("1".equals(status)) {
- return "等級(jí)1";
- } else if ("2".equals(status)) {
- return "等級(jí)2";
- } else if ("3".equals(status)) {
- return "等級(jí)3";
- }
- }
- return "未知";
- }
- }
發(fā)現(xiàn)雖然依然會(huì)有 (false,等級(jí)4,3) 這樣的錯(cuò)誤撤回?cái)?shù)據(jù)(這是 udf 決定的，沒(méi)法避免)，但是我們可以發(fā)現(xiàn)最終的結(jié)果是 (true,等級(jí)2,4)，結(jié)果依然是正確的。
再來(lái)分析下問(wèn)什么這種方式可以解決，如圖 plan。
6
發(fā)現(xiàn)映射 udf 算子所處的位置不在兩個(gè) GroupAggregrate 之間了，因此在 retract 消息發(fā)送之后，不會(huì)被映射到錯(cuò)誤其他 key 中，因此所有的 retract 消息都會(huì)正常處理。
7.展望篇-有什么機(jī)制可以根本避免這種情況
可以將「撤回消息(做減法)」，**最新的結(jié)果消息(做加法)**做成一個(gè)原子消息從上游發(fā)給下游，下游統(tǒng)一進(jìn)行原子性處理，關(guān)聯(lián) udf 時(shí)，也只對(duì) group key 關(guān)聯(lián)一次即可。
