場(chǎng)景描述

Kafka Source 接收并處理來(lái)自 Kafka 的點(diǎn)擊數(shù)據(jù)(指定事件時(shí)間)，開(kāi)一個(gè)滾動(dòng)窗口(Tumble Windows) 每 10 秒統(tǒng)計(jì)一次 pv 并將結(jié)果輸出到 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;

運(yùn)行以上的 FlinkSQL 后觀察發(fā)現(xiàn)數(shù)據(jù)流正常，但是一直沒(méi)能觸發(fā)窗口計(jì)算，沒(méi)有結(jié)果輸出。

watermark

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

并行流中的 watermark

在 多并行度的情況下，source 的每個(gè) sub task 通常獨(dú)立生成水印。watermark 通過(guò) operator 時(shí)會(huì)推進(jìn) operators 處的當(dāng)前 event time，同時(shí) operators 會(huì)為下游生成一個(gè)新的 watermark。多并行度的情況下 watermark對(duì)齊會(huì)取所有 channel 最小的 watermark。

并行流中的 watermark

原因分析

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

解決方案

可通過(guò)手動(dòng)設(shè)置并行度來(lái)解決，保證 source 并行度 <= 目標(biāo) Topic Partition 分區(qū)數(shù)。這里將 source 并行度設(shè)置為 1 之后便可正常輸出結(jié)果。

并行度設(shè)置為 1

正常輸出