Flink 并行流中 watermark 機制無法觸發窗口計算的原因分析
場景描述
Kafka Source 接收并處理來自 Kafka 的點擊數據(指定事件時間)，開一個滾動窗口(Tumble Windows) 每 10 秒統計一次 pv 并將結果輸出到 Print Sink 中。
CREATE TABLE sourceTable (
message STRING,
time_ltz AS TO_TIMESTAMP_LTZ(CAST(JSON_VALUE(JSON_VALUE(message, '$.request_body'),'$.clickTime') AS INTEGER),0),
WATERMARK FOR time_ltz AS time_ltz - INTERVAL '3' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'matrix_json_click_log_test',
'properties.bootstrap.servers' = 'xxxxxxxxx:9527',
'properties.group.id' = 'flinkTestGroup',
'scan.startup.mode' = 'latest-offset',
'format' = 'json'
);
CREATE TABLE sinkTable (
window_start TIMESTAMP(3),
window_end TIMESTAMP(3),
click_pv BIGINT
)
WITH (
'connector' = 'print'
);
INSERT INTO sinkTable(
window_start,
window_end,
click_pv
)
select window_start,window_end,COUNT(*)
FROM TABLE (
TUMBLE( TABLE sourceTable, DESCRIPTOR(time_ltz) , INTERVAL '10' SECOND))
GROUP BY window_start, window_end;
運行以上的 FlinkSQL 后觀察發現數據流正常，但是一直沒能觸發窗口計算，沒有結果輸出。
watermark
watermark 是用于處理亂序事件的。流處理從事件產生，到流經source，再到operator，中間是有一個過程和時間的。雖然大部分情況下流到 operator 的數據都是按照事件產生的時間順序來的，但是也不排除由于網絡等原因導致部分數據延遲到達，產生亂序。對于遲到的數據我們又不能無限期地等待下去，因此需要有個衡量事件時間進度的機制來保證一個特定的時間后必須觸發 window 進行計算，這個特別的機制就是 watermark。
并行流中的 watermark
在 多并行度的情況下，source 的每個 sub task 通常獨立生成水印。watermark 通過 operator 時會推進 operators 處的當前 event time，同時 operators 會為下游生成一個新的 watermark。多并行度的情況下 watermark對齊會取所有 channel 最小的 watermark。
并行流中的 watermark
原因分析
由于目標 Topic 只有一個分區而 source 并行度設置為 2 ，這導致了只有一個線程可以處理該分區而另一個線程獲取不到數據，因此一直沒能獲取最小的 watermark。最終導致一直無法觸發窗口計算。
解決方案
可通過手動設置并行度來解決，保證 source 并行度 <= 目標 Topic Partition 分區數。這里將 source 并行度設置為 1 之后便可正常輸出結果。
并行度設置為 1
正常輸出
