簡介

在實時數據開發過程中，大家經常會用 Flink SQL 或者 Flink DataStream API 來做數據加工。通常情況下選用2者都能加工出想要的數據，但是總會有 Flink SQL 覆蓋不了的問題，但 SQL 的易用性又難以讓人釋懷。所以有些場景在使用 FLink SQL 開始就與需要額外注意，下面就介紹一種多表關聯時存在部分列更新（partial Update）場景，在 DataStream API 和 Flink SQL 開發時都容易忽視的情況而導致的問題。為了簡化問題描述，采用了Flink SQL 來闡述此類問題。

場景介紹

多表關聯時表 A 關聯表 B, 表 A 具有pk1, field1, field2, field3字段，表 B 具有 pk2, field4, field5, field6 字段，表 A 通過 pk1 關聯表B pk2。使用 Flink SQL 會如下實現：

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;

上述實例中有明顯特征：使用了Join 關聯, 且需要注意的是寫入的數據庫 sink 是 StarRocks。StarRocks 存在如下特性：當表是主鍵表時是不支持部分列更新（ Partial Update）的，實際上大部分時候大家都用的是主鍵表。

然后在一個SQL查詢數據的接口就遇到了如下問題：每次從接口查詢返回的結果都不穩定，同樣的查詢條件不同時機返回的結果不一樣。SQL查詢語句如下：

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

為什么SQL查詢的結果會不一致呢？起初排查原因發現 group by 返回結果有多條，而在SQL 中也沒有使用 order by 對數據進行排序，所以導致了結果不穩定。后又排查為什么會出現多條結果呢？于是懷疑 field1, field2, field3 有不符合預期的數據。如： 

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

其中第2條是多余的，不應該出現。結果發現可能是如下原因導致的：這3個字段 filed1, field2, filed3 在StarRocks數據庫中會一直在變化，不停的寫入新值。導致 SQL 查詢時可以查到 field3 為 null 的數據。
為什么field3為不斷變化呢？究其原因是：StarRocks 主鍵表不支持部分列更新（Partial Update）。當field3 為null時，同樣會被寫入 StarRocks。我們在通過JDQ讀取表A field1, field2, field3 數據給表C寫入數據時，當JDQ 消息隊列中表A的記錄存在亂序場景且field3 字段可能為null時，最終寫入StarRocks的field3 字段會出現時而為null，時而不為null。 所以SQL查詢接口中 group by的結果會出現不穩定。

總結

為什么在開發的時候當時沒有發現 StarRocks 主鍵表這個問題呢？原因：1. 大家所關注的部分列更新，多數是關注insert into table_C(field1, field2, field3) 中不包含的字段field4,field5...等被更新為null，而當前場景是會把 field3 為null的值也寫入SR數據庫中，這不是我們期望的結果。2.表A作為主表，通常不會出現開始field3有值后來又沒有值（null）的場景。出現這個現象大概率是因為上游JDQ消息隊列中的數據亂序了，導致field3 為null的后出現了。而這種問題又比較難發現。

什么情況下會出現此類問題呢？寫入的數據庫不支持部分列更新場景時會出現。如StarRocks, Doris。因為MySQL, ES，ClickHouse的部分表引擎支持部分列更新，所以在MySQL, ES，ClickHouse中不會出現。

同理在 DataStream API 中如果表 A,表 B 關聯后的數據直接寫入StarRocks 的話，也會出現此類問題。
以上這個問題在 Flink SQL 中無法解決，在 Flink DataStream API 中可以模擬部分列更新來避免此類問題。具體方法：在DatStream 任務中增加一個MapState, 用來在新數據到來時從MapState拿出緩存的數據，并和新到來的數據進行合并，來實現部分列更新功能，最后再寫入 StarRocks。
雖然問題不是Flink SQL導致的，但是上面的問題可以通過Flink DataStream API來規避。

審核編輯 黃宇