日本综合一区二区|亚洲中文天堂综合|日韩欧美自拍一区|男女精品天堂一区|欧美自拍第6页亚洲成人精品一区|亚洲黄色天堂一区二区成人|超碰91偷拍第一页|日韩av夜夜嗨中文字幕|久久蜜综合视频官网|精美人妻一区二区三区

RELATEED CONSULTING
相關(guān)咨詢
選擇下列產(chǎn)品馬上在線溝通
服務(wù)時(shí)間：8:30-17:00
你可能遇到了下面的問題
關(guān)閉右側(cè)工具欄

新聞中心

這里有您想知道的互聯(lián)網(wǎng)營銷解決方案
FlinkSQL知其所以然：萬字詳述FlinkSQL四種時(shí)間窗口語義！

DML：窗口聚合

大家好我是老羊，由于窗口涉及到的知識內(nèi)容比較多，所以博主先為大家說明介紹下面內(nèi)容時(shí)的思路，大家跟著思路走。思路如下：

創(chuàng)新互聯(lián)公司專注于云州企業(yè)網(wǎng)站建設(shè),成都響應(yīng)式網(wǎng)站建設(shè)公司,商城建設(shè)。云州網(wǎng)站建設(shè)公司,為云州等地區(qū)提供建站服務(wù)。全流程按需規(guī)劃網(wǎng)站，專業(yè)設(shè)計(jì)，全程項(xiàng)目跟蹤，創(chuàng)新互聯(lián)公司專業(yè)和態(tài)度為您提供的服務(wù)

  • 先介紹 Flink SQL 支持的 4 種時(shí)間窗口。
  • 分別詳細(xì)介紹上述的 4 種時(shí)間窗口的功能及 SQL 語法。
  • 結(jié)合實(shí)際案例介紹 4 種時(shí)間窗口。

首先來看看 Flink SQL 中支持的 4 種窗口的運(yùn)算。

  • 滾動窗口(TUMBLE)。
  • 滑動窗口(HOP)。
  • Session 窗口(SESSION)。
  • 漸進(jìn)式窗口(CUMULATE)。

1、滾動窗口(TUMBLE)

滾動窗口定義：滾動窗口將每個(gè)元素指定給指定窗口大小的窗口。滾動窗口具有固定大小，且不重疊。例如，指定一個(gè)大小為 5 分鐘的滾動窗口。在這種情況下，F(xiàn)link 將每隔 5 分鐘開啟一個(gè)新的窗口，其中每一條數(shù)都會劃分到唯一一個(gè) 5 分鐘的窗口中，如下圖所示。

tumble window

應(yīng)用場景：常見的按照一分鐘對數(shù)據(jù)進(jìn)行聚合，計(jì)算一分鐘內(nèi) PV，UV 數(shù)據(jù)。

實(shí)際案例：簡單且常見的分維度分鐘級別同時(shí)在線用戶數(shù)、總銷售額。

那么上面這個(gè)案例的 SQL 要咋寫呢?

關(guān)于滾動窗口，在 1.13 版本之前和 1.13 及之后版本有兩種 Flink SQL 實(shí)現(xiàn)方式，分別是：

  • Group Window Aggregation(1.13 之前只有此類方案，此方案在 1.13 及之后版本已經(jīng)標(biāo)記為廢棄，不推薦小伙伴萌使用)。
  • ? Windowing TVF(1.13 及之后建議使用 Windowing TVF)。

博主這里兩種方法都會介紹：

  • Group Window Aggregation 方案(支持 Batch\Streaming 任務(wù))：
-- 數(shù)據(jù)源表
CREATE TABLE source_table (
    -- 維度數(shù)據(jù)
    dim STRING,
    -- 用戶 id
    user_id BIGINT,
    -- 用戶
    price BIGINT,
    -- 事件時(shí)間戳
    row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),
    -- watermark 設(shè)置
    WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '10',
  'fields.dim.length' = '1',
  'fields.user_id.min' = '1',
  'fields.user_id.max' = '100000',
  'fields.price.min' = '1',
  'fields.price.max' = '100000'
)
-- 數(shù)據(jù)匯表
CREATE TABLE sink_table (
    dim STRING,
    pv BIGINT,
    sum_price BIGINT,
    max_price BIGINT,
    min_price BIGINT,
    uv BIGINT,
    window_start bigint
) WITH (
  'connector' = 'print'
)
-- 數(shù)據(jù)處理邏輯
insert into sink_table
select 
    dim,
    count(*) as pv,
    sum(price) as sum_price,
    max(price) as max_price,
    min(price) as min_price,
    -- 計(jì)算 uv 數(shù)
    count(distinct user_id) as uv,
    UNIX_TIMESTAMP(CAST(tumble_start(row_time, interval '1' minute) AS STRING)) * 1000  as window_start
from source_table
group by
    dim,
    tumble(row_time, interval '1' minute)

可以看到 Group Window Aggregation 滾動窗口的 SQL 語法就是把 tumble window 的聲明寫在了 group by 子句中，即 tumble(row_time, interval '1' minute)，第一個(gè)參數(shù)為事件時(shí)間的時(shí)間戳;第二個(gè)參數(shù)為滾動窗口大小。

  • Window TVF 方案(1.13 只支持 Streaming 任務(wù))：
-- 數(shù)據(jù)源表
CREATE TABLE source_table (
    -- 維度數(shù)據(jù)
    dim STRING,
    -- 用戶 id
    user_id BIGINT,
    -- 用戶
    price BIGINT,
    -- 事件時(shí)間戳
    row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),
    -- watermark 設(shè)置
    WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '10',
  'fields.dim.length' = '1',
  'fields.user_id.min' = '1',
  'fields.user_id.max' = '100000',
  'fields.price.min' = '1',
  'fields.price.max' = '100000'
)
-- 數(shù)據(jù)匯表
CREATE TABLE sink_table (
    dim STRING,
    pv BIGINT,
    sum_price BIGINT,
    max_price BIGINT,
    min_price BIGINT,
    uv BIGINT,
    window_start bigint
) WITH (
  'connector' = 'print'
)
-- 數(shù)據(jù)處理邏輯
insert into sink_table
SELECT 
    dim,
    UNIX_TIMESTAMP(CAST(window_start AS STRING)) * 1000 as window_start,
    count(*) as pv,
    sum(price) as sum_price,
    max(price) as max_price,
    min(price) as min_price,
    count(distinct user_id) as uv
FROM TABLE(TUMBLE(
        TABLE source_table
        , DESCRIPTOR(row_time)
        , INTERVAL '60' SECOND))
GROUP BY window_start, 
      window_end,
      dim

可以看到 Windowing TVF 滾動窗口的寫法就是把 tumble window 的聲明寫在了數(shù)據(jù)源的 Table 子句中，即 TABLE(TUMBLE(TABLE source_table, DESCRIPTOR(row_time), INTERVAL '60' SECOND))，包含三部分參數(shù)。

第一個(gè)參數(shù) TABLE source_table 聲明數(shù)據(jù)源表;第二個(gè)參數(shù) DESCRIPTOR(row_time) 聲明數(shù)據(jù)源的時(shí)間戳;第三個(gè)參數(shù) INTERVAL '60' SECOND 聲明滾動窗口大小為 1 min。

SQL 語義：

由于離線沒有相同的時(shí)間窗口聚合概念，這里就直接說實(shí)時(shí)場景 SQL 語義，假設(shè) Orders 為 kafka，target_table 也為 Kafka，這個(gè) SQL 生成的實(shí)時(shí)任務(wù)，在執(zhí)行時(shí)，會生成三個(gè)算子：

數(shù)據(jù)源算子(From Order)：連接到 Kafka topic，數(shù)據(jù)源算子一直運(yùn)行，實(shí)時(shí)的從 Order Kafka 中一條一條的讀取數(shù)據(jù)，然后一條一條發(fā)送給下游的 窗口聚合算子。

窗口聚合算子(TUMBLE 算子)：接收到上游算子發(fā)的一條一條的數(shù)據(jù)，然后將每一條數(shù)據(jù)按照時(shí)間戳劃分到對應(yīng)的窗口中(根據(jù)事件時(shí)間、處理時(shí)間的不同語義進(jìn)行劃分)，上述案例為事件時(shí)間，事件時(shí)間中，滾動窗口算子接收到上游的 Watermark 大于窗口的結(jié)束時(shí)間時(shí)，則說明當(dāng)前這一分鐘的滾動窗口已經(jīng)結(jié)束了，將窗口計(jì)算完的結(jié)果發(fā)往下游算子(一條一條發(fā)給下游 數(shù)據(jù)匯算子)。

數(shù)據(jù)匯算子(INSERT INTO target_table)：接收到上游發(fā)的一條一條的數(shù)據(jù)，寫入到 target_table Kafka 中。

這個(gè)實(shí)時(shí)任務(wù)也是 24 小時(shí)一直在運(yùn)行的，所有的算子在同一時(shí)刻都是處于 running 狀態(tài)的。

注意：

事件時(shí)間中滾動窗口的窗口計(jì)算觸發(fā)是由 Watermark 推動的。

2、滑動窗口(HOP)

滑動窗口定義：滑動窗口也是將元素指定給固定長度的窗口。與滾動窗口功能一樣，也有窗口大小的概念。不一樣的地方在于，滑動窗口有另一個(gè)參數(shù)控制窗口計(jì)算的頻率(滑動窗口滑動的步長)。因此，如果滑動的步長小于窗口大小，則滑動窗口之間每個(gè)窗口是可以重疊。在這種情況下，一條數(shù)據(jù)就會分配到多個(gè)窗口當(dāng)中。舉例，有 10 分鐘大小的窗口，滑動步長為 5 分鐘。這樣，每 5 分鐘會劃分一次窗口，這個(gè)窗口包含的數(shù)據(jù)是過去 10 分鐘內(nèi)的數(shù)據(jù)，如下圖所示。

hop window

應(yīng)用場景：比如計(jì)算同時(shí)在線的數(shù)據(jù)，要求結(jié)果的輸出頻率是 1 分鐘一次，每次計(jì)算的數(shù)據(jù)是過去 5 分鐘的數(shù)據(jù)(有的場景下用戶可能在線，但是可能會 2 分鐘不活躍，但是這也要算在同時(shí)在線數(shù)據(jù)中，所以取最近 5 分鐘的數(shù)據(jù)就能計(jì)算進(jìn)去了)。

實(shí)際案例：簡單且常見的分維度分鐘級別同時(shí)在線用戶數(shù)，1 分鐘輸出一次，計(jì)算最近 5 分鐘的數(shù)據(jù)。

依然是 Group Window Aggregation、Windowing TVF 兩種方案：

  • Group Window Aggregation 方案(支持 Batch\Streaming 任務(wù))：
-- 數(shù)據(jù)源表
CREATE TABLE source_table (
    -- 維度數(shù)據(jù)
    dim STRING,
    -- 用戶 id
    user_id BIGINT,
    -- 用戶
    price BIGINT,
    -- 事件時(shí)間戳
    row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),
    -- watermark 設(shè)置
    WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '10',
  'fields.dim.length' = '1',
  'fields.user_id.min' = '1',
  'fields.user_id.max' = '100000',
  'fields.price.min' = '1',
  'fields.price.max' = '100000'
);
-- 數(shù)據(jù)匯表
CREATE TABLE sink_table (
    dim STRING,
    uv BIGINT,
    window_start bigint
) WITH (
  'connector' = 'print'
);
-- 數(shù)據(jù)處理邏輯
insert into sink_table
SELECT dim,
    UNIX_TIMESTAMP(CAST(hop_start(row_time, interval '1' minute, interval '5' minute) AS STRING)) * 1000 as window_start, 
    count(distinct user_id) as uv
FROM source_table
GROUP BY dim
    , hop(row_time, interval '1' minute, interval '5' minute)

可以看到 Group Window Aggregation 滾動窗口的寫法就是把 hop window 的聲明寫在了 group by 子句中，即 hop(row_time, interval '1' minute, interval '5' minute)。其中：

第一個(gè)參數(shù)為事件時(shí)間的時(shí)間戳;第二個(gè)參數(shù)為滑動窗口的滑動步長;第三個(gè)參數(shù)為滑動窗口大小。

  • Windowing TVF 方案(1.13 只支持 Streaming 任務(wù))：
-- 數(shù)據(jù)源表
CREATE TABLE source_table (
    -- 維度數(shù)據(jù)
    dim STRING,
    -- 用戶 id
    user_id BIGINT,
    -- 用戶
    price BIGINT,
    -- 事件時(shí)間戳
    row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),
    -- watermark 設(shè)置
    WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '10',
  'fields.dim.length' = '1',
  'fields.user_id.min' = '1',
  'fields.user_id.max' = '100000',
  'fields.price.min' = '1',
  'fields.price.max' = '100000'
);
-- 數(shù)據(jù)匯表
CREATE TABLE sink_table (
    dim STRING,
    uv BIGINT,
    window_start bigint
) WITH (
  'connector' = 'print'
);
-- 數(shù)據(jù)處理邏輯
insert into sink_table
SELECT 
    dim,
    UNIX_TIMESTAMP(CAST(window_start AS STRING)) * 1000 as window_start, 
    count(distinct user_id) as bucket_uv
FROM TABLE(HOP(
        TABLE source_table
        , DESCRIPTOR(row_time)
        , INTERVAL '1' MINUTES, INTERVAL '5' MINUTES))
GROUP BY window_start, 
      window_end,
      dim

可以看到 Windowing TVF 滾動窗口的寫法就是把 hop window 的聲明寫在了數(shù)據(jù)源的 Table 子句中，即 TABLE(HOP(TABLE source_table, DESCRIPTOR(row_time), INTERVAL '1' MINUTES, INTERVAL '5' MINUTES))，包含四部分參數(shù)：

第一個(gè)參數(shù) TABLE source_table 聲明數(shù)據(jù)源表;第二個(gè)參數(shù) DESCRIPTOR(row_time) 聲明數(shù)據(jù)源的時(shí)間戳;第三個(gè)參數(shù) INTERVAL '1' MINUTES 聲明滾動窗口滑動步長大小為 1 min。第四個(gè)參數(shù) INTERVAL '5' MINUTES 聲明滾動窗口大小為 5 min。

SQL 語義：

滑動窗口語義和滾動窗口類似，這里不再贅述。

3、Session 窗口(SESSION)

  • Session 窗口定義：Session 時(shí)間窗口和滾動、滑動窗口不一樣，其沒有固定的持續(xù)時(shí)間，如果在定義的間隔期(Session Gap)內(nèi)沒有新的數(shù)據(jù)出現(xiàn)，則 Session 就會窗口關(guān)閉。如下圖對比所示：

session window

實(shí)際案例：計(jì)算每個(gè)用戶在活躍期間(一個(gè) Session)總共購買的商品數(shù)量，如果用戶 5 分鐘沒有活動則視為 Session 斷開。

目前 1.13 版本中 Flink SQL 不支持 Session 窗口的 Window TVF，所以這里就只介紹 Group Window Aggregation 方案：

  • Group Window Aggregation 方案(支持 Batch\Streaming 任務(wù))：
-- 數(shù)據(jù)源表，用戶購買行為記錄表
CREATE TABLE source_table (
    -- 維度數(shù)據(jù)
    dim STRING,
    -- 用戶 id
    user_id BIGINT,
    -- 用戶
    price BIGINT,
    -- 事件時(shí)間戳
    row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),
    -- watermark 設(shè)置
    WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '10',
  'fields.dim.length' = '1',
  'fields.user_id.min' = '1',
  'fields.user_id.max' = '100000',
  'fields.price.min' = '1',
  'fields.price.max' = '100000'
);
-- 數(shù)據(jù)匯表
CREATE TABLE sink_table (
    dim STRING,
    pv BIGINT, -- 購買商品數(shù)量
    window_start bigint
) WITH (
  'connector' = 'print'
);
-- 數(shù)據(jù)處理邏輯
insert into sink_table
SELECT 
    dim,
    UNIX_TIMESTAMP(CAST(session_start(row_time, interval '5' minute) AS STRING)) * 1000 as window_start, 
    count(1) as pv
FROM source_table
GROUP BY dim
      , session(row_time, interval '5' minute)

注意：

上述 SQL 任務(wù)是在整個(gè) Session 窗口結(jié)束之后才會把數(shù)據(jù)輸出。Session 窗口即支持 處理時(shí)間 也支持 事件時(shí)間。但是處理時(shí)間只支持在Streaming 任務(wù)中運(yùn)行，Batch 任務(wù)不支持。

可以看到 Group Window Aggregation 中 Session 窗口的寫法就是把 session window 的聲明寫在了 group by 子句中，即 session(row_time, interval '5' minute)。其中：

第一個(gè)參數(shù)為事件時(shí)間的時(shí)間戳;第二個(gè)參數(shù)為 Session gap 間隔。

SQL 語義：

Session 窗口語義和滾動窗口類似，這里不再贅述。

4、漸進(jìn)式窗口(CUMULATE)

漸進(jìn)式窗口定義(1.13 只支持 Streaming 任務(wù))：漸進(jìn)式窗口在其實(shí)就是 固定窗口間隔內(nèi)提前觸發(fā)的的滾動窗口，其實(shí)就是 Tumble Window + early-fire 的一個(gè)事件時(shí)間的版本。例如，從每日零點(diǎn)到當(dāng)前這一分鐘繪制累積 UV，其中 10:00 時(shí)的 UV 表示從 00:00 到 10:00 的 UV總數(shù)。漸進(jìn)式窗口可以認(rèn)為是首先開一個(gè)最大窗口大小的滾動窗口，然后根據(jù)用戶設(shè)置的觸發(fā)的時(shí)間間隔將這個(gè)滾動窗口拆分為多個(gè)窗口，這些窗口具有相同的窗口起點(diǎn)和不同的窗口終點(diǎn)。如下圖所示：

cumulate window

應(yīng)用場景：周期內(nèi)累計(jì) PV，UV 指標(biāo)(如每天累計(jì)到當(dāng)前這一分鐘的 PV，UV)。這類指標(biāo)是一段周期內(nèi)的累計(jì)狀態(tài)，對分析師來說更具統(tǒng)計(jì)分析價(jià)值，而且?guī)缀跛械膹?fù)合指標(biāo)都是基于此類指標(biāo)的統(tǒng)計(jì)(不然離線為啥都要累計(jì)一天的數(shù)據(jù)，而不要一分鐘累計(jì)的數(shù)據(jù)呢)。

實(shí)際案例：每天的截止當(dāng)前分鐘的累計(jì) money(sum(money))，去重 id 數(shù)(count(distinct id))。每天代表漸進(jìn)式窗口大小為 1 天，分鐘代表漸進(jìn)式窗口移動步長為分鐘級別。舉例如下：

明細(xì)輸入數(shù)據(jù)：

預(yù)期經(jīng)過漸進(jìn)式窗口計(jì)算的輸出數(shù)據(jù)：

轉(zhuǎn)化為折線圖長這樣：

當(dāng)日累計(jì)

可以看到，其特點(diǎn)就在于，每一分鐘的輸出結(jié)果都是當(dāng)天零點(diǎn)累計(jì)到當(dāng)前的結(jié)果。

漸進(jìn)式窗口目前只有 Windowing TVF 方案支持：

  • Windowing TVF 方案(1.13 只支持 Streaming 任務(wù))：
-- 數(shù)據(jù)源表
CREATE TABLE source_table (
    -- 用戶 id
    user_id BIGINT,
    -- 用戶
    money BIGINT,
    -- 事件時(shí)間戳
    row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),
    -- watermark 設(shè)置
    WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '10',
  'fields.user_id.min' = '1',
  'fields.user_id.max' = '100000',
  'fields.price.min' = '1',
  'fields.price.max' = '100000'
);
-- 數(shù)據(jù)匯表
CREATE TABLE sink_table (
    window_end bigint,
    window_start bigint,
    sum_money BIGINT,
    count_distinct_id bigint
) WITH (
  'connector' = 'print'
);
-- 數(shù)據(jù)處理邏輯
insert into sink_table
SELECT 
    UNIX_TIMESTAMP(CAST(window_end AS STRING)) * 1000 as window_end, 
    window_start, 
    sum(money) as sum_money,
    count(distinct id) as count_distinct_id
FROM TABLE(CUMULATE(
       TABLE source_table
       , DESCRIPTOR(row_time)
       , INTERVAL '60' SECOND
       , INTERVAL '1' DAY))
GROUP BY
    window_start, 
    window_end

可以看到 Windowing TVF 滾動窗口的寫法就是把 cumulate window 的聲明寫在了數(shù)據(jù)源的 Table 子句中，即 TABLE(CUMULATE(TABLE source_table, DESCRIPTOR(row_time), INTERVAL '60' SECOND, INTERVAL '1' DAY))，其中包含四部分參數(shù)：

第一個(gè)參數(shù) TABLE source_table 聲明數(shù)據(jù)源表;第二個(gè)參數(shù) DESCRIPTOR(row_time) 聲明數(shù)據(jù)源的時(shí)間戳;第三個(gè)參數(shù) INTERVAL '60' SECOND 聲明漸進(jìn)式窗口觸發(fā)的漸進(jìn)步長為 1 min。第四個(gè)參數(shù) INTERVAL '1' DAY 聲明整個(gè)漸進(jìn)式窗口的大小為 1 天，到了第二天新開一個(gè)窗口重新累計(jì)。

SQL 語義：

漸進(jìn)式窗口語義和滾動窗口類似，這里不再贅述。

5、Window TVF 支持 Grouping Sets、Rollup、Cube

具體應(yīng)用場景：實(shí)際的案例場景中，經(jīng)常會有多個(gè)維度進(jìn)行組合(cube)計(jì)算指標(biāo)的場景。如果把每個(gè)維度組合的代碼寫一遍，然后

union all 起來，這樣寫起來非常麻煩，而且會導(dǎo)致一個(gè)數(shù)據(jù)源讀取多遍。

這時(shí)，有離線 Hive SQL 使用經(jīng)驗(yàn)的小伙伴萌就會想到，如果有了 Grouping Sets，我們就可以直接用 Grouping Sets 將維度組合寫在一條 SQL 中，寫起來方便并且執(zhí)行效率也高。當(dāng)然，F(xiàn)link 支持這個(gè)功能。

但是目前 Grouping Sets 只在 Window TVF 中支持，不支持 Group Window Aggregation。

來一個(gè)實(shí)際案例感受一下，計(jì)算每日零點(diǎn)累計(jì)到當(dāng)前這一分鐘的分匯總、age、sex、age+sex 維度的用戶數(shù)。

-- 用戶訪問明細(xì)表
CREATE TABLE source_table (
    age STRING,
    sex STRING,
    user_id BIGINT,
    row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),
    WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '1',
  'fields.age.length' = '1',
  'fields.sex.length' = '1',
  'fields.user_id.min' = '1',
  'fields.user_id.max' = '100000'
);
CREATE TABLE sink_table (
    age STRING,
    sex STRING,
    uv BIGINT,
    window_end bigint
) WITH (
  'connector' = 'print'
);
insert into sink_table
SELECT 
    UNIX_TIMESTAMP(CAST(window_end AS STRING)) * 1000 as window_end, 
    if (age is null, 'ALL', age) as age,
    if (sex is null, 'ALL', sex) as sex,
    count(distinct user_id) as bucket_uv
FROM TABLE(CUMULATE(
       TABLE source_table
       , DESCRIPTOR(row_time)
       , INTERVAL '5' SECOND
       , INTERVAL '1' DAY))
GROUP BY 
    window_start, 
    window_end,
    -- grouping sets 寫法
    GROUPING SETS (
        ()
        , (age)
        , (sex)
        , (age, sex)
    )

小伙伴萌這里需要注意下!

Flink SQL 中 Grouping Sets 的語法和 Hive SQL 的語法有一些不同，如果我們使用 Hive SQL 實(shí)現(xiàn)上述 SQL 的語義，其實(shí)現(xiàn)如下：

insert into sink_table
SELECT 
    UNIX_TIMESTAMP(CAST(window_end AS STRING)) * 1000 as window_end, 
    if (age is null, 'ALL', age) as age,
    if (sex is null, 'ALL', sex) as sex,
    count(distinct user_id) as bucket_uv
FROM source_table
GROUP BY
    age
    , sex
-- hive sql grouping sets 寫法
GROUPING SETS (
    ()
    , (age)
    , (sex)
    , (age, sex)
)

當(dāng)前名稱：FlinkSQL知其所以然：萬字詳述FlinkSQL四種時(shí)間窗口語義！
文章出自：http://www.dlmjj.cn/article/dhhjihe.html