場景描述

Kafka Source 接收并處理來自 Kafka 的點擊數(shù)據(jù)(指定事件時間)，開一個滾動窗口(Tumble Windows) 每 10 秒統(tǒng)計一次 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 后觀察發(fā)現(xiàn)數(shù)據(jù)流正常，但是一直沒能觸發(fā)窗口計算，沒有結果輸出。

watermark

watermark 是用于處理亂序事件的。流處理從事件產(chǎn)生，到流經(jīng)source，再到operator，中間是有一個過程和時間的。雖然大部分情況下流到 operator 的數(shù)據(jù)都是按照事件產(chǎn)生的時間順序來的，但是也不排除由于網(wǎng)絡等原因導致部分數(shù)據(jù)延遲到達，產(chǎn)生亂序。對于遲到的數(shù)據(jù)我們又不能無限期地等待下去，因此需要有個衡量事件時間進度的機制來保證一個特定的時間后必須觸發(fā) window 進行計算，這個特別的機制就是 watermark。

并行流中的 watermark

在 多并行度的情況下，source 的每個 sub task 通常獨立生成水印。watermark 通過 operator 時會推進 operators 處的當前 event time，同時 operators 會為下游生成一個新的 watermark。多并行度的情況下 watermark對齊會取所有 channel 最小的 watermark。

并行流中的 watermark

原因分析

由于目標 Topic 只有一個分區(qū)而 source 并行度設置為 2 ，這導致了只有一個線程可以處理該分區(qū)而另一個線程獲取不到數(shù)據(jù)，因此一直沒能獲取最小的 watermark。最終導致一直無法觸發(fā)窗口計算。

解決方案

可通過手動設置并行度來解決，保證 source 并行度 <= 目標 Topic Partition 分區(qū)數(shù)。這里將 source 并行度設置為 1 之后便可正常輸出結果。

并行度設置為 1

正常輸出