#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);
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);
© 版权声明
文章版权归作者所有,未经允许请勿转载。