簡(jiǎn)介

在實(shí)時(shí)數(shù)據(jù)開(kāi)發(fā)過(guò)程中，大家經(jīng)常會(huì)用 Flink SQL 或者 Flink DataStream API 來(lái)做數(shù)據(jù)加工。通常情況下選用2者都能加工出想要的數(shù)據(jù)，但是總會(huì)有 Flink SQL 覆蓋不了的問(wèn)題，但 SQL 的易用性又難以讓人釋?xiě)?。所以有些?chǎng)景在使用 FLink SQL 開(kāi)始就與需要額外注意，下面就介紹一種多表關(guān)聯(lián)時(shí)存在部分列更新（partial Update）場(chǎng)景，在 DataStream API 和 Flink SQL 開(kāi)發(fā)時(shí)都容易忽視的情況而導(dǎo)致的問(wèn)題。為了簡(jiǎn)化問(wèn)題描述，采用了Flink SQL 來(lái)闡述此類(lèi)問(wèn)題。

場(chǎng)景介紹

多表關(guān)聯(lián)時(shí)表 A 關(guān)聯(lián)表 B, 表 A 具有pk1, field1, field2, field3字段，表 B 具有 pk2, field4, field5, field6 字段，表 A 通過(guò) pk1 關(guān)聯(lián)表B pk2。使用 Flink SQL 會(huì)如下實(shí)現(xiàn)：

CREATE TABLE jdq_source(
pk1 INT,
field1 STIRNG,
field2 STIRNG,
field3 STIRNG,
PRIMARY KEY(pk1) NOT ENFORCED
) WITH(...);

CREATE TABLE sr_sink(
pk1 INT,
field1 STRING,
field2 STRING,
field3 STRING,
field4 STRING,
field5 STRING,
field6 STRING,
PRIMARY KEY(pk2) NOT ENFORCED
) WITH (...);

INSERT INTO C
SELECT A.pk1,A.field1,A.field2,A.field3,B.pk2,B.field4,B.field5,B.field6 FROM jdq_source A
INNER JOIN sr_sink B
ON A.pk1 = B.pk2;

上述實(shí)例中有明顯特征：使用了Join 關(guān)聯(lián), 且需要注意的是寫(xiě)入的數(shù)據(jù)庫(kù) sink 是 StarRocks。StarRocks 存在如下特性：當(dāng)表是主鍵表時(shí)是不支持部分列更新（ Partial Update）的，實(shí)際上大部分時(shí)候大家都用的是主鍵表。

然后在一個(gè)SQL查詢(xún)數(shù)據(jù)的接口就遇到了如下問(wèn)題：每次從接口查詢(xún)返回的結(jié)果都不穩(wěn)定，同樣的查詢(xún)條件不同時(shí)機(jī)返回的結(jié)果不一樣。SQL查詢(xún)語(yǔ)句如下：

select C.field1,C.field2,C.field3
FROM C group by field1,field2,field3;

為什么SQL查詢(xún)的結(jié)果會(huì)不一致呢？起初排查原因發(fā)現(xiàn) group by 返回結(jié)果有多條，而在SQL 中也沒(méi)有使用 order by 對(duì)數(shù)據(jù)進(jìn)行排序，所以導(dǎo)致了結(jié)果不穩(wěn)定。后又排查為什么會(huì)出現(xiàn)多條結(jié)果呢？于是懷疑 field1, field2, field3 有不符合預(yù)期的數(shù)據(jù)。如： 

20240530, 2, 3
20240530, 2, null
20240531, 2, 4

其中第2條是多余的，不應(yīng)該出現(xiàn)。結(jié)果發(fā)現(xiàn)可能是如下原因?qū)е碌模哼@3個(gè)字段 filed1, field2, filed3 在StarRocks數(shù)據(jù)庫(kù)中會(huì)一直在變化，不停的寫(xiě)入新值。導(dǎo)致 SQL 查詢(xún)時(shí)可以查到 field3 為 null 的數(shù)據(jù)。
為什么field3為不斷變化呢？究其原因是：StarRocks 主鍵表不支持部分列更新（Partial Update）。當(dāng)field3 為null時(shí)，同樣會(huì)被寫(xiě)入 StarRocks。我們?cè)谕ㄟ^(guò)JDQ讀取表A field1, field2, field3 數(shù)據(jù)給表C寫(xiě)入數(shù)據(jù)時(shí)，當(dāng)JDQ 消息隊(duì)列中表A的記錄存在亂序場(chǎng)景且field3 字段可能為null時(shí)，最終寫(xiě)入StarRocks的field3 字段會(huì)出現(xiàn)時(shí)而為null，時(shí)而不為null。 所以SQL查詢(xún)接口中 group by的結(jié)果會(huì)出現(xiàn)不穩(wěn)定。

總結(jié)

為什么在開(kāi)發(fā)的時(shí)候當(dāng)時(shí)沒(méi)有發(fā)現(xiàn) StarRocks 主鍵表這個(gè)問(wèn)題呢？原因：1. 大家所關(guān)注的部分列更新，多數(shù)是關(guān)注insert into table_C(field1, field2, field3) 中不包含的字段field4,field5...等被更新為null，而當(dāng)前場(chǎng)景是會(huì)把 field3 為null的值也寫(xiě)入SR數(shù)據(jù)庫(kù)中，這不是我們期望的結(jié)果。2.表A作為主表，通常不會(huì)出現(xiàn)開(kāi)始field3有值后來(lái)又沒(méi)有值（null）的場(chǎng)景。出現(xiàn)這個(gè)現(xiàn)象大概率是因?yàn)樯嫌蜫DQ消息隊(duì)列中的數(shù)據(jù)亂序了，導(dǎo)致field3 為null的后出現(xiàn)了。而這種問(wèn)題又比較難發(fā)現(xiàn)。

什么情況下會(huì)出現(xiàn)此類(lèi)問(wèn)題呢？寫(xiě)入的數(shù)據(jù)庫(kù)不支持部分列更新場(chǎng)景時(shí)會(huì)出現(xiàn)。如StarRocks, Doris。因?yàn)镸ySQL, ES，ClickHouse的部分表引擎支持部分列更新，所以在MySQL, ES，ClickHouse中不會(huì)出現(xiàn)。

同理在 DataStream API 中如果表 A,表 B 關(guān)聯(lián)后的數(shù)據(jù)直接寫(xiě)入StarRocks 的話(huà)，也會(huì)出現(xiàn)此類(lèi)問(wèn)題。
以上這個(gè)問(wèn)題在 Flink SQL 中無(wú)法解決，在 Flink DataStream API 中可以模擬部分列更新來(lái)避免此類(lèi)問(wèn)題。具體方法：在DatStream 任務(wù)中增加一個(gè)MapState, 用來(lái)在新數(shù)據(jù)到來(lái)時(shí)從MapState拿出緩存的數(shù)據(jù)，并和新到來(lái)的數(shù)據(jù)進(jìn)行合并，來(lái)實(shí)現(xiàn)部分列更新功能，最后再寫(xiě)入 StarRocks。
雖然問(wèn)題不是Flink SQL導(dǎo)致的，但是上面的問(wèn)題可以通過(guò)Flink DataStream API來(lái)規(guī)避。

審核編輯 黃宇