關于 Flink Regular Join 與 TTL 的理解
作者：SapphireCoder
在 Regular Join 時 Flink 會將兩條沒有時間窗口限制的流的所有數據存儲在 State 中，由于流是無窮無盡持續流入的，隨著時間的不斷推進，內存中積累的狀態會越來越多。
對于流查詢，Regular Join 的語法是最靈活的，它允許任何類型的更新(插入、更新、刪除)輸入表。
Regular Join 包含以下幾種（以 L 作為左流中的數據標識，R 作為右流中的數據標識）：
- Inner Join（Inner Equal Join）：當兩條流 Join 到才會輸出 +[L, R]
- Left Join（Outer Equal Join）：左流數據到達之后 Join 到 R 流數據則輸出 +[L, R]，沒 Join 到輸出 +[L, null]）。如果右流之后數據到達之后，發現左流之前輸出過沒有 Join 到的數據，則會發起回撤流，先輸出 -[L, null]，然后輸出 +[L, R]。
- Right Join（Outer Equal Join）：與 Left Join 邏輯相反。
- Full Join（Outer Equal Join）：流任務中，左流或者右流的數據到達之后，無論有沒有 Join 到另外一條流的數據，都會輸出（對右流來說：Join 到輸出 +[L, R]，沒 Join 到輸出 +[null, R]；對左流來說：Join 到輸出 +[L, R]，沒 Join 到輸出 +[L, null]）。如果一條流的數據到達之后，發現之前另一條流之前輸出過沒有 Join 到的數據，則會發起回撤流（左流數據到達為例：回撤 -[null, R]，輸出 +[L, R]，右流數據到達為例：回撤 -[L, null]，輸出 +[L, R]）。
Regular Inner Join
Flink SQL?：
CREATE TABLE matchResult (
guid STRING
) WITH (
'connector' = 'kafka',
'topic' = 'match_result_log_test',
'properties.bootstrap.servers' = 'xxxxxxxxxxxxxxxxxxx',
'properties.group.id' = 'flinkTestGroup',
'scan.startup.mode' = 'latest-offset',
'format' = 'json'
);
CREATE TABLE readRecord (
guid STRING,
book_name STRING
) WITH (
'connector' = 'kafka',
'topic' = 'read_record_log_test',
'properties.bootstrap.servers' = 'xxxxxxxxxxxxxxxxxxx',
'properties.group.id' = 'flinkTestGroup',
'scan.startup.mode' = 'latest-offset',
'format' = 'json'
);
CREATE TABLE sink_table (
guid STRING,
book_name STRING
) WITH (
'connector' = 'print'
);
INSERT INTO sink_table
SELECT
matchResult.guid,
readRecord.book_name
FROM matchResult
INNER JOIN readRecord ON matchResult.guid = readRecord.guid;
輸出結果解析?：
-- L 流數據達到，由于沒有 Join 到 R 流數據而且是 inner join 便不輸出結果
+I[111, book1] -- R 流數據達到， Join 到 L 流數據，便輸出 +I[111, book1]
-- R 流數據達到，由于沒有 Join 到 L 流數據而且是 inner join 便不輸出結果
+I[222, book2] -- L 流數據達到， Join 到 R 流數據便輸出結果
Regular Left Join（Right join 則相反）
Flink SQL：
CREATE TABLE matchResult (
guid STRING
) WITH (
'connector' = 'kafka',
'topic' = 'match_result_log_test',
'properties.bootstrap.servers' = 'xxxxxxxxxxxxxxxxxxx',
'properties.group.id' = 'flinkTestGroup',
'scan.startup.mode' = 'latest-offset',
'format' = 'json'
);
CREATE TABLE readRecord (
guid STRING,
book_name STRING
) WITH (
'connector' = 'kafka',
'topic' = 'read_record_log_test',
'properties.bootstrap.servers' = 'xxxxxxxxxxxxxxxxxxx',
'properties.group.id' = 'flinkTestGroup',
'scan.startup.mode' = 'latest-offset',
'format' = 'json'
);
CREATE TABLE sink_table (
guid STRING,
book_name STRING
) WITH (
'connector' = 'print'
);
INSERT INTO sink_table
SELECT
matchResult.guid,
readRecord.book_name
FROM matchResult
LEFT JOIN readRecord ON matchResult.guid = readRecord.guid;
輸出結果解析：
+I[111, null] -- L 流數據達到，沒有 Join 到 R 流數據，便輸出 +[L, null]
-D[111, null] -- R 流的數據到達，發現 L 流之前輸出過沒有 Join 到的數據，則會發起回撤流，先輸出 -[L, null]
+I[111, book1] -- 再輸出 +[L, R]
-- 這里模擬一條 R 流 guid = 222 的數據到達，由于是 left join 且沒有 join 到 L 流，因此不做輸出
+I[222, book2] -- 當 L 流 guid = 222 的數據達到 join R 流 后輸出結果 +[L, R]
Regular Full Join
Flink SQL：
CREATE TABLE matchResult (
guid STRING
) WITH (
'connector' = 'kafka',
'topic' = 'match_result_log_test',
'properties.bootstrap.servers' = 'xxxxxxxxxxxxxxxxxxx',
'properties.group.id' = 'flinkTestGroup',
'scan.startup.mode' = 'latest-offset',
'format' = 'json'
);
CREATE TABLE readRecord (
guid STRING,
book_name STRING
) WITH (
'connector' = 'kafka',
'topic' = 'read_record_log_test',
'properties.bootstrap.servers' = 'xxxxxxxxxxxxxxxxxxx',
'properties.group.id' = 'flinkTestGroup',
'scan.startup.mode' = 'latest-offset',
'format' = 'json'
);
CREATE TABLE sink_table (
guid STRING,
book_name STRING
) WITH (
'connector' = 'print'
);
INSERT INTO sink_table
SELECT
matchResult.guid,
readRecord.book_name
FROM matchResult
FULL JOIN readRecord ON matchResult.guid = readRecord.guid;
輸出結果解析：
+I[111, null] -- L 流數據達到，沒有 Join 到 R 流數據，便輸出 +I[L, null]
+I[null, book2] -- R 流數據達到，沒有 Join 到 R 流數據，便輸出 +I[null, R]
-D[null, book2] -- L 流新數據到達，發現之前 R 流之前輸出過沒有 Join 到的數據，則發起回撤流，先輸出 -D[null, R]
+I[222, book2] -- 再輸出 +I[L, R]
-D[111, null] -- 反之同理
+I[111, book1]
TTL 概念
在 Regular Join 時 Flink 會將兩條沒有時間窗口限制的流的所有數據存儲在 State 中，由于流是無窮無盡持續流入的，隨著時間的不斷推進，內存中積累的狀態會越來越多。
針對這個問題，Flink 提出了空閑狀態保留時間（Idle State Retention Time）的概念。通過為每個狀態設置 Timer，如果這個狀態中途被訪問過，則重新設置 Timer；否則（如果狀態一直未被訪問，長期處于 Idle 狀態）則在 Timer 到期時做狀態清理。這樣，就可以確保每個狀態都能得到及時的清理，可以通過 table.exec.state.ttl 參數進行控制（注意：這同時也會對結果的準確性有所影響，因此需要合理的權衡）。
責任編輯：趙寧寧
來源：
今日頭條
