MySQL Flink Watermark实现事件时间处理的关键技术_Mysql

MySQL Flink Watermark实现事件时间处理的关键技术

2023-05-09 14:04奔跑如风 Mysql

Flink Watermark是Flink流处理框架中实现事件时间处理的关键技术之一。它是通过一种机制来解决数据乱序和延迟的问题，使得Flink可以在处理迟到的数据时保证结果的准确性。Flink Watermark可以在数据流中插入一个时间戳，将数据流转化

1.概述

生活中有种场景：

车辆进入隧道，信号不好，出了隧道后，信号就正常了。

正常情况下，车辆进入隧道后，如果车辆正常，没有事故，会正常驶出隧道。

在正常的隧道行驶过程中，可能会因为信号的原因，导致数据没有像信号正常的时候那么快到达。

也就是说，这种情况下，数据出现了延迟。我们把这种延迟数据称之为迟到数据。

生活中，这种场景非常多，比如：车辆进入地下车库，手机欠费，网络抖动等。这都属于生活的正常情况。无法避免。

程序中，一般不会允许数据丢失。所以，我们程序会推出一些机制来保证迟到数据被正常处理。

Watermark就是用来保证正常迟到的数据被正确的处理。

Watermark，也叫水印，或者是水位线。用来处理一定程度下的延迟数据。

2.SQL案例-演示Watermark为零的情况

				?

									#1.创建表

									CREATE TABLE source_table ( 

									 user_id STRING, 

									 price BIGINT,

									 `timestamp` bigint,

									 row_time AS TO_TIMESTAMP(FROM_UNIXTIME(`timestamp`)),

									 watermark for row_time as row_time - interval '0' second

									) WITH (

									  'connector' = 'socket',

									  'hostname' = 'node1', 

									  'port' = '9999',

									  'format' = 'csv'

									);

									#2.数据查询SQL

									select

									user_id,

									count(*) as pv,

									sum(price) as sum_price,

									UNIX_TIMESTAMP(CAST(tumble_start(row_time, interval '5' second) AS STRING)) * 1000  as window_start,

									UNIX_TIMESTAMP(CAST(tumble_end(row_time, interval '5' second) AS STRING)) * 1000  as window_end

									from source_table

									group by

									    user_id,

									    tumble(row_time, interval '5' second);

3.SQL案例-演示Watermark不为零的情况

Watermark不为零，就有可能是两种情况：

小于0，窗口会提前触发计算，这种情况在实际应用不存在，所以这里也不讨论
大于0，窗口会延迟触发计算，延迟的时间就是我们设置的Watermark的值

这里，我们主要是讨论Watermark>0的情况。

				?

									#1.创建表

									CREATE TABLE source_table ( 

									 user_id STRING, 

									 price BIGINT,

									 `timestamp` bigint,

									 row_time AS TO_TIMESTAMP(FROM_UNIXTIME(`timestamp`)),

									 watermark for row_time as row_time - interval '2' second

									) WITH (

									  'connector' = 'socket',

									  'hostname' = 'node1', 

									  'port' = '9999',

									  'format' = 'csv'

									);

									#2.Watermark的解释

									WATERMARK FOR ts AS ts - INTERVAL '2' SECOND

									这里的2，表示，数据允许延迟2秒钟到达，窗口会在(正常结束+延迟时间)后触发计算

									#3.查询SQL

									select

									user_id,

									count(*) as pv,

									sum(price) as sum_price,

									UNIX_TIMESTAMP(CAST(tumble_start(row_time, interval '5' second) AS STRING)) * 1000  as window_start,

									UNIX_TIMESTAMP(CAST(tumble_end(row_time, interval '5' second) AS STRING)) * 1000  as window_end

									from source_table

									group by

									    user_id,

									    tumble(row_time, interval '5' second);