Re:回复：flink 1.12如何实现window topN功能

Tue, 22 Jun 2021 21:08:07 -0700 

-- source
CREATE TABLE tmall_item(
itemID VARCHAR,
itemType VARCHAR,
onSellTime TIMESTAMP(3),
price DOUBLE,
proctime AS PROCTIME(),
WATERMARK FOR onSellTime AS onSellTime - INTERVAL '5' SECOND
) WITH (
'connector' = 'filesystem' ,
'path' = 'file:///path/to/over-window.csv',
'format' = 'csv'
);

-- sink
CREATE TABLE print_table (
itemID VARCHAR,
itemType VARCHAR,
onSellTime TIMESTAMP(3),
price DOUBLE
) WITH (
'connector' = 'print'
);

-- insert
INSERT INTO print_table
SELECT itemID,
itemType,
onSellTime,
price
FROM (
SELECT itemID,
itemType,
onSellTime,
price,
ROW_NUMBER() OVER (
PARTITION BY itemID, DATE_FORMAT(proctime, 'yyyyMMddHHmm')
ORDER BY onSellTime DESC
) AS row_num
FROM tmall_item
     ) WHERE row_num = 1;

















在 2021-06-23 11:06:39，"杨光跃" <[email protected]> 写道：
>应该是这样吧
>
>
>1. 第一步以主键group by 以及分时间窗口
>SELECT 主键, TUMBLE_START(ts, INTERVAL '10' SECOND) as wStart, FROM source_event 
> group by TUMBLE(ts, INTERVAL '10' SECOND), 主键
>
>
>2. 根据上一步的结果取top5
>select * from (select * ,ROW_NUMBER() OVER (PARTITION BY wStart ORDER BY 
>处理时间字段 ) as rownum from 上一步的虚拟表) where rownum <= 5
>
>| |
>杨光跃
>|
>|
>[email protected]
>|
>签名由网易邮箱大师定制
>在2021年6月23日 10:58，casel.chen<[email protected]> 写道：
>你指的是TopN吗？https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/sql/queries.html#top-n
>但我想知道window topN写法，跟这个还不一样。
>我的需求是：
>cdc场景同一个主键数据变更频繁，我想定义一个5秒处理时间窗口，在该窗口内取同一主键最新变更记录。用flink sql 1.12如何实现？
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>在 2021-06-23 10:18:01，"杨光跃" <[email protected]> 写道：
>
>
>Apache Flink 1.12 Documentation: Queries
>| |
>杨光跃
>|
>|
>[email protected]
>|
>签名由网易邮箱大师定制
>在2021年6月23日 10:09，casel.chen<[email protected]> 写道：
>请不要截图哦
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>
>在 2021-06-23 09:47:46，"杨光跃" <[email protected]> 写道：
>
>1.12也支持的
>| |
>杨光跃
>|
>|
>[email protected]
>|
>签名由网易邮箱大师定制
>在2021年6月23日 09:45，casel.chen<[email protected]> 写道：
>官网文档上显示flink 1.13支持window topN，但flink 1.12没看到相关文档，请问有什么work around方法吗？

回复