Flink基礎（61）：FLINK SQL(38) 視窗函式（2）滾動視窗

阿新 • • 發佈：2021-08-07

本文為您介紹如何使用實時計算Flink版滾動視窗函式。

1 定義

滾動視窗（TUMBLE）將每個元素分配到一個指定大小的視窗中。通常，滾動視窗有一個固定的大小，並且不會出現重疊。例如，如果指定了一個5分鐘大小的滾動視窗，無限流的資料會根據時間劃分為[0:00, 0:05)、[0:05, 0:10)、[0:10, 0:15)等視窗。下圖展示了一個30秒的滾動視窗。

2 語法

TUMBLE函式用在GROUP BY子句中，用來定義滾動視窗。

TUMBLE(<time-attr>, <size-interval>)
<size-interval>: INTERVAL 'string' timeUnit

說明<time-attr>引數必須是時間流中的一個合法的時間屬性欄位，指定為Processing Time或Event Time，請參見概述，瞭解如何定義時間屬性和Watermark。

3 標識函式

使用標識函式選出視窗的起始時間或者結束時間，視窗的時間屬性用於下級Window的聚合。


視窗標識函式	返回型別	描述
`TUMBLE_START(time-attr, size-interval)`	TIMESTAMP	返回視窗的起始時間（包含邊界）。例如`[00:10,00:15)`視窗，返回`00:10`。
`TUMBLE_END(time-attr, size-interval)`	TIMESTAMP	返回視窗的結束時間（包含邊界）。例如`[00:00, 00:15]`視窗，返回`00:15`。
`TUMBLE_ROWTIME(time-attr, size-interval)`	TIMESTAMP(rowtime-attr)	返回視窗的結束時間（不包含邊界）。例如`[00:00, 00:15]`視窗，返回`00:14:59.999`。返回值是一個rowtime attribute，即可以基於該欄位做時間屬性的操作，例如，級聯視窗只能用在基於Event Time的Window上，詳情請參見級聯視窗。
`TUMBLE_PROCTIME(time-attr, size-interval)`	TIMESTAMP(rowtime-attr)	返回視窗的結束時間（不包含邊界）。例如`[00:00, 00:15]` 視窗，返回`00:14:59.999`。返回值是一個Proctime Attribute，即可以基於該欄位做時間屬性的操作。例如，級聯視窗只能用在基於Processing Time的Window上，詳情請參見級聯視窗。

4 使用Event Time統計每個使用者每分鐘在指定網站的單擊數示例

測試資料


username（VARCHAR）	click_url（VARCHAR）	ts（TIMESTAMP）
Jark	`http://taobao.com/xxx`	`2017-10-10 10:00:00.0`
Jark	`http://taobao.com/xxx`	`2017-10-10 10:00:10.0`
Jark	`http://taobao.com/xxx`	`2017-10-10 10:00:49.0`
Jark	`http://taobao.com/xxx`	`2017-10-10 10:01:05.0`
Jark	`http://taobao.com/xxx`	`2017-10-10 10:01:58.0`
Timo	`http://taobao.com/xxx`	`2017-10-10 10:02:10.0`

測試語句

CREATE TABLE user_clicks(
  username varchar,
  click_url varchar,
  ts timeStamp,
  WATERMARK wk FOR ts as withOffset(ts, 2000) --為rowtime定義Watermark。
) with (
  type='datahub',
  ...
);

CREATE TABLE tumble_output(
  window_start TIMESTAMP,
  window_end TIMESTAMP,
  username VARCHAR,
  clicks BIGINT
) with (
  type='RDS'
);

INSERT INTO tumble_output
SELECT
TUMBLE_START(ts, INTERVAL '1' MINUTE) as window_start,
TUMBLE_END(ts, INTERVAL '1' MINUTE) as window_end,
username,
COUNT(click_url)
FROM user_clicks
GROUP BY TUMBLE(ts, INTERVAL '1' MINUTE), username;

測試結果


window_start （TIMESTAMP）	window_end （TIMESTAMP）	username（VARCHAR）	clicks（BIGINT）
`2017-10-10 10:00:00.0`	`2017-10-10 10:01:00.0`	Jark	3
`2017-10-10 10:01:00.0`	`2017-10-10 10:02:00.0`	Jark	2
`2017-10-10 10:02:00.0`	`2017-10-10 10:03:00.0`	Timo	1

5 使用Processing Time統計每個使用者每分鐘在指定網站的單擊數示例

測試資料


username （VARCHAR）	click_url（VARCHAR）
Jark	`http://taobao.com/xxx`
Jark	`http://taobao.com/xxx`
Jark	`http://taobao.com/xxx`
Jark	`http://taobao.com/xxx`
Jark	`http://taobao.com/xxx`
Timo	`http://taobao.com/xxx`

測試語句

CREATE TABLE window_test (
  username   VARCHAR,
  click_url  VARCHAR,
  ts as PROCTIME()
) WITH (
  type='datahub',
  ...
);

CREATE TABLE tumble_output(
  window_start TIMESTAMP,
  window_end TIMESTAMP,
  username VARCHAR,
  clicks BIGINT
) with (
  type='print'
);

INSERT INTO tumble_output
SELECT
TUMBLE_START(ts, INTERVAL '1' MINUTE),
TUMBLE_END(ts, INTERVAL '1' MINUTE),
username,
COUNT(click_url)
FROM window_test
GROUP BY TUMBLE(ts, INTERVAL '1' MINUTE), username;