建站
咨詢
售后
建站咨詢
新聞中心
場(chǎng)景描述
Kafka Source 接收并處理來自 Kafka 的點(diǎn)擊數(shù)據(jù)(指定事件時(shí)間)，開一個(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ù)流正常，但是一直沒能觸發(fā)窗口計(jì)算，沒有結(jié)果輸出。
watermark
watermark 是用于處理亂序事件的。流處理從事件產(chǎn)生，到流經(jīng)source，再到operator，中間是有一個(gè)過程和時(shí)間的。雖然大部分情況下流到 operator 的數(shù)據(jù)都是按照事件產(chǎn)生的時(shí)間順序來的，但是也不排除由于網(wǎng)絡(luò)等原因?qū)е虏糠謹(jǐn)?shù)據(jù)延遲到達(dá)，產(chǎn)生亂序。對(duì)于遲到的數(shù)據(jù)我們又不能無限期地等待下去，因此需要有個(gè)衡量事件時(shí)間進(jìn)度的機(jī)制來保證一個(gè)特定的時(shí)間后必須觸發(fā) window 進(jìn)行計(jì)算，這個(gè)特別的機(jī)制就是 watermark。
并行流中的 watermark
在 多并行度的情況下，source 的每個(gè) sub task 通常獨(dú)立生成水印。watermark 通過 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ù)，因此一直沒能獲取最小的 watermark。最終導(dǎo)致一直無法觸發(fā)窗口計(jì)算。
解決方案
可通過手動(dòng)設(shè)置并行度來解決，保證 source 并行度 <= 目標(biāo) Topic Partition 分區(qū)數(shù)。這里將 source 并行度設(shè)置為 1 之后便可正常輸出結(jié)果。
并行度設(shè)置為 1
正常輸出
分享題目：Flink 并行流中 watermark 機(jī)制無法觸發(fā)窗口計(jì)算的原因分析
網(wǎng)站路徑：http://www.dlmjj.cn/article/djpdege.html
