建站
咨詢
售后
建站咨詢
新聞中心
這里有您想知道的互聯(lián)網營銷解決方案
關于FlinkRegularJoin與TTL的理解
對于流查詢，Regular Join 的語法是最靈活的，它允許任何類型的更新(插入、更新、刪除)輸入表。
網站建設哪家好，找成都創(chuàng)新互聯(lián)公司！專注于網頁設計、網站建設、微信開發(fā)、成都小程序開發(fā)、集團企業(yè)網站建設等服務項目。為回饋新老客戶創(chuàng)新互聯(lián)還提供了武勝免費建站歡迎大家使用！
Regular Join 包含以下幾種（以 L 作為左流中的數據標識，R 作為右流中的數據標識）：
- Inner Join（Inner Equal Join）：當兩條流 Join 到才會輸出 +[L, R]
- Left Join（Outer Equal Join）：左流數據到達之后 Join 到 R 流數據則輸出 +[L, R]，沒 Join 到輸出 +[L, null]）。如果右流之后數據到達之后，發(fā)現左流之前輸出過沒有 Join 到的數據，則會發(fā)起回撤流，先輸出 -[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]）。如果一條流的數據到達之后，發(fā)現之前另一條流之前輸出過沒有 Join 到的數據，則會發(fā)起回撤流（左流數據到達為例：回撤 -[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 流的數據到達，發(fā)現 L 流之前輸出過沒有 Join 到的數據，則會發(fā)起回撤流，先輸出 -[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 流新數據到達，發(fā)現之前 R 流之前輸出過沒有 Join 到的數據，則發(fā)起回撤流，先輸出 -D[null, R]
+I[222, book2] -- 再輸出 +I[L, R]
-D[111, null] -- 反之同理
+I[111, book1]
TTL 概念
在 Regular Join 時 Flink 會將兩條沒有時間窗口限制的流的所有數據存儲在 State 中，由于流是無窮無盡持續(xù)流入的，隨著時間的不斷推進，內存中積累的狀態(tài)會越來越多。
針對這個問題，Flink 提出了空閑狀態(tài)保留時間（Idle State Retention Time）的概念。通過為每個狀態(tài)設置 Timer，如果這個狀態(tài)中途被訪問過，則重新設置 Timer；否則（如果狀態(tài)一直未被訪問，長期處于 Idle 狀態(tài)）則在 Timer 到期時做狀態(tài)清理。這樣，就可以確保每個狀態(tài)都能得到及時的清理，可以通過 table.exec.state.ttl 參數進行控制（注意：這同時也會對結果的準確性有所影響，因此需要合理的權衡）。
文章題目：關于FlinkRegularJoin與TTL的理解
分享鏈接：http://fisionsoft.com.cn/article/dhoghsj.html
