flink sql 知其是以然（八）：flink sql tumble window 的奇妙解析之路

1.序篇-本文結構

針對 datastream api 大家都比較熟悉了，還是那句話，在 datastream 中，你寫的代碼邏輯是什麼樣的，它最終的執行方式就是什麼樣的。

但是對于 flink sql 的執行過程，大家還是不熟悉的。上節使用 ETL，group agg（sum，count等）簡單聚合類 query 帶大家走進一條 flink sql query 邏輯的世界。幫大家至少能夠熟悉在 flink sql 程式運作時知道 flink 程式在幹什麼。

此節就是視窗聚合章節的第一篇，以一個最簡單、最常用的分鐘 tumble window 聚合案例給大家介紹其使用方式和原理。

由于 flink 1.13 引入了 window tvf，是以 1.13 和 1.12 及之前版本的實作不同。本節先介紹 flink 1.12 及之前的 tumble window 實作。這也是大家在引入 flink sql 能力時最常使用的。

本節依然從以下幾個章節給大家詳細介紹 flink sql 的能力。

1. 目标篇-本文能幫助大家了解 flink sql 什麼？
   • 回顧上節的 flink sql 适用場景的結論
2. 概念篇-先聊聊常見的視窗聚合
   • 視窗竟然拖慢資料産出？
   • 常用的視窗
3. 實戰篇-簡單的 tumble window 案例和運作原理
   • 先看一個 datastream 視窗案例
   • flink sql tumble window 的語義
   • tumble window 實際案例
   • GeneratedWatermarkGenerator - flink 1.12.1
   • BinaryRowDataKeySelector - flink 1.12.1
   • AggregateWindowOperator - flink 1.12.1
4. 總結與展望篇

先說說結論，以下這些結論已經在上節說過了，此處附上上節文章：

1. 場景問題：flink sql 很适合簡單 ETL，以及基本全部場景下的聚合類名額（本節要介紹的 tumble window 就在聚合類名額的範疇之内）。
2. 文法問題：flink sql 文法其實是和其他 sql 文法基本一緻的。基本不會産生文法問題阻礙使用 flink sql。但是本節要介紹的 tumble window 的文法就是略有不同的那部分。下面詳細介紹。
3. 運作問題：檢視 flink sql 任務時的一些技巧，以及其中一些可能會碰到的坑：
   • 去 flink webui 就能看到這個任務目前在做什麼。包括算子名稱都會給直接展示給我們目前哪個算子在幹啥事情，在處理啥邏輯。
   • sql 的 watermark 類型要設定為 TIMESTAMP(3)。
   • 事件時間邏輯中，sql api 和 datastream api 對于資料記錄時間戳存儲邏輯是不一樣的。datastream api：每條記錄的 rowtime 是放在 StreamRecord 中的時間戳字段中的。sql api：時間戳是每次都從資料中進行擷取的。算子中會維護一個下标。可以按照下标從資料中擷取時間戳。

2.目标篇-本文能幫助大家了解 flink sql tumble window 什麼？

關于 flink sql tumble window 一般都會有以下問題。本文的目标也是為大家解答這些問題：

1. 場景問題：場景問題就不必多說，datastream 在 tumble window 場景下的應用很多了，分鐘級别聚合等常用場景
2. 文法問題：flink sql 寫 tumble window 任務時是一種與 hive sql 中沒有的文法。下文詳細介紹。
3. 運作問題：使用一條簡單的 tumble window sql 幫大家從 transformation、runtime 幫大家了解 tumble window 的整體運作機制。
4. 了解誤區：既然是 sql 必然要遵循 sql 語義，sql tumble window 聚合是輸入多條，産出一條資料。并不像 datastream 那樣可以在視窗 udf 中做到多對多。

在正式開始聊 tumble window 之前，先看看上節 flink sql 适用場景的結論。讓大家先有 flink sql 的一個整體印象以及結論。

2.1.回顧上節的 flink sql 适用場景的結論

不裝了，我坦白了，flink sql 其實很适合幹的活就是 dwd 清洗，dws 聚合。

此處主要針對實時數倉的場景來說。flink sql 能幹 dwd 清洗，dws 聚合，基本上實時數倉的大多數場景都能給覆寫了。

flink sql 牛逼！！！

但是！！！

經過部落客使用 flink sql 經驗來看，并不是所有的 dwd，dws 聚合場景都适合 flink sql（截止發文階段來說）！！！

其實這些目前不适合 flink sql 的場景總結下來就是在處理上比 datastream 還是會有一定的損失。

先總結下使用場景：

1. dwd：簡單的清洗、複雜的清洗、次元的擴充、各種 udf 的使用

2. dws：各類聚合

然後分适合的場景和不适合的場景來說，因為隻這一篇不能覆寫所有的内容，是以本文此處先大緻給個結論，之後會結合具體的場景較長的描述。

• 适合的場景：
  1. 簡單的 dwd 清洗場景
  2. 全場景的 dws 聚合場景
• 目前不太适合的場景：
  1. 複雜的 dwd 清洗場景：舉例比如使用了很多 udf 清洗，尤其是使用很多的 json 類解析清洗
  2. 關聯次元場景：舉例比如 datastream 中經常會有攢一批資料批量通路外部接口的場景，flink sql 目前對于這種場景雖然有 localcache、異步通路能力，但是依然還是一條一條通路外部緩存，這樣相比批量通路還是會有性能差距。

3.概念篇-先聊聊常見的視窗聚合

視窗聚合大家都在 datastream api 中很熟悉了，目前在實時資料處理的過程中，視窗計算可以說是最重要、最常用的一種計算方式了。

但是在抛出視窗概念之前，部落客有幾個關于視窗的小想法說一下。

3.1.視窗竟然拖慢資料産出？

一個小想法。

先抛結論：視窗會拖慢實時資料的産出，是在目前下遊分析引擎能力有限的情況下的一種妥協方案。

站在資料開發以及需求方的世界中，當然希望所有的資料都是實時來的，實時處理的，實時産出的，實時展現的。

舉個例子：如果你要滿足一個一分鐘視窗聚合的 pv，uv，或者其他聚合需求。

olap 資料服務引擎 就可以滿足上述的實時來的，實時處理的，實時産出的，實時展現的的場景。flink 消費處理明細資料，産出到 kafka，然後直接導入到 olap 引擎中。查詢時直接用 olap 做聚合。這其中是沒有任何視窗的概念的。但是整個鍊路中，要保障端對端精确一次，要保障大資料量情況下 olap 引擎能夠秒級查詢傳回，更何況有一些去重類名額的計算，等等場景。把這些壓力都放在 olap 引擎的壓力是很大的。

是以在 flink 資料計算引擎中就誕生了視窗的概念。我們可以直接在計算引擎中進行視窗聚合計算，然後等到視窗結束之後直接把結果資料産出。這就出現了部落客所說的視窗拖慢了實時資料産出的情況。而且視窗在處理不好的情況下可能會導緻資料丢失。

關于上述兩種情況的具體優劣選擇，都由大家自行選擇。上述隻是引出部落客一些想法。

3.2.常用的視窗

目前已知的視窗分為以下四種。

1. Tumble Windows2. Hop Windows3. Cumulate Windows4. Session Windows

這些視窗的具體描述直接見官網，有詳細的說明。此處不贅述。

https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/table/sql/queries/window-agg/

此處介紹下 flink 中常常會涉及到的兩個容易混淆的概念就是：視窗 + key。這裡來形象的說明下。

• 視窗：時間周期上面的劃分。将無限流進行縱向切分，将無限流切分為一個一個的視窗，視窗相當于是無限流中的一段時間内的資料。
• key：資料類别上面的劃分。将無限流進行橫向劃分，相同 key 的資料會被劃分到一組中，這個 key 的資料也是一條無限流。

如下圖所示。

1

4.實戰篇-簡單的 tumble window 案例和運作原理

源碼公衆号背景回複flink sql tumble window 的奇妙解析之路擷取。

4.1.先看一個 datastream 視窗案例

在介紹 sql tumble window 視窗算子執行案例之前，先看一個 datastream 中的視窗算子案例。其邏輯都是相通的。會對我們了解 sql tumble window 算子有幫助。

我們先看看 datastream 處理邏輯。

以下面這個為例。

public class _04_TumbleWindowTest {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        env.setParallelism(1);

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        env.addSource(new UserDefinedSource())
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple4<String, String, Integer, Long>>(Time.seconds(0)) {
                    @Override
                    public long extractTimestamp(Tuple4<String, String, Integer, Long> element) {
                        return element.f3;
                    }
                })
                .keyBy(new KeySelector<Tuple4<String, String, Integer, Long>, String>() {
                    @Override
                    public String getKey(Tuple4<String, String, Integer, Long> row) throws Exception {
                        return row.f0;
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .sum(2)
                .print();

        env.execute("1.12.1 DataStream TUMBLE WINDOW 案例");
    }

    private static class UserDefinedSource implements SourceFunction<Tuple4<String, String, Integer, Long>> {

        private volatile boolean isCancel;

        @Override
        public void run(SourceContext<Tuple4<String, String, Integer, Long>> sourceContext) throws Exception {

            while (!this.isCancel) {

                sourceContext.collect(Tuple4.of("a", "b", 1, System.currentTimeMillis()));

                Thread.sleep(10L);
            }

        }

        @Override
        public void cancel() {
            this.isCancel = true;
        }
    }
}
           

複制

datastream 生産的具體的 transformation 如下圖：

24

其中我們隻關注最重要的

WindowOperator

算子。

25

其中

WindowOperator

算子包含的重要屬性如下圖。

26

來看看

WindowOperator

的執行邏輯。視窗執行的整體詳細流程可以參考：http://wuchong.me/blog/2016/05/25/flink-internals-window-mechanism/

23

4.2.flink sql tumble window 的語義

介紹到 tumble window 的語義，總要有對比的去介紹。這裡的參照物就是 datastream api。

在 datastream api 中。tumble window 一般用作以下兩種場景。

1. 業務場景：使用 tumble window 很輕松的計算出視窗内的聚合資料。一般是多條輸入資料，視窗結束時一條輸出資料。
2. 優化場景：視窗聚合一批資料然後批量通路外部存儲擴充次元、或者有一些自定義的處理邏輯。一般是多條輸入資料，視窗結束時多條輸出資料。

但是在 sql api 中。tumble window 是聚合（group by）語義，聚合在 sql 标準中的資料處理邏輯是多條輸入，在視窗觸發時就輸出一條資料的語義。而上面的常常用在 datastream 中的優化場景是多對多的場景。是以和 sql 語義不符合。是以 flink sql tumble window 一般都是用于計算聚合運算值來使用。

4.3.tumble window 實際案例

滾動視窗的特性就是會将無限流進行縱向劃分成一個一個的視窗，每個視窗都是相同的大小，并且不重疊。

22

本文主要介紹 flink 1.12 及之前版本的實作。下一篇文章介紹 flink 1.13 的實作。

來，在介紹原理之前，總要先用起來，我們就以下面這個例子展開。

1.（flink 1.12.1）場景：簡單且常見的分次元分鐘級别同時線上使用者數、總銷售額

資料源表：

CREATE TABLE source_table (
    -- 次元資料
    dim STRING,
    -- 使用者 id
    user_id BIGINT,
    -- 使用者
    price BIGINT,
    -- 事件時間戳
    row_time AS cast(CURRENT_TIMESTAMP as timestamp(3)),
    -- watermark 設定
    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'
)
           

複制

Notes - 關于 watermark 容易踩得坑：sql 的 watermark 類型必須要設定為

TIMESTAMP(3)

。

資料彙表：

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'
)
           

複制

資料處理邏輯：

可以看下下面文法，視窗聚合的寫法有專門的

tumble(row_time, interval '1' minute)

寫法，這就是與平常我們寫的 hive sql，mysql 等不一樣的地方。

insert into sink_table
select dim,
       sum(bucket_pv) as pv,
       sum(bucket_sum_price) as sum_price,
       max(bucket_max_price) as max_price,
       min(bucket_min_price) as min_price,
       sum(bucket_uv) as uv,
       max(window_start) as window_start
from (
     select dim,
            count(*) as bucket_pv,
            sum(price) as bucket_sum_price,
            max(price) as bucket_max_price,
            min(price) as bucket_min_price,
            -- 計算 uv 數
            count(distinct user_id) as bucket_uv,
            cast(tumble_start(row_time, interval '1' minute) as bigint) * 1000 as window_start
     from source_table
     group by
            -- 按照使用者 id 進行分桶，防止資料傾斜
            mod(user_id, 1024),
            dim,
            tumble(row_time, interval '1' minute)
)
group by dim,
         window_start
           

複制

2.運作：可以看到，其實在 flink sql 任務中，其會把對應的處理邏輯給寫到算子名稱上面。

Notes - 觀察 flink sql 技巧 1：這個其實就是我們觀察 flink sql 任務的第一個技巧。如果你想知道你的 flink 任務在幹啥，第一反應是去 flink webui 看看這個任務目前在做什麼。包括算子名稱都會給直接展示給我們目前哪個算子在幹啥事情，在處理啥邏輯

先看一下整個算子圖，如下圖。從左到右總共分為三個算子。

1. 第一個算子就是資料源算子
2. 第二個算子就是分了桶的視窗聚合算子，第一個算子和第二個算子之間 hash 傳輸就是按照 group key 進行 hash 傳輸
3. 第三個算子就是外層進行合桶計算的算子，同樣也是 hash 傳輸，将分桶的資料在一個算子中進行合并計算

5

來看看每一個算子具體做了什麼事情。

第一個算子：

1. table scan 讀取資料源
2. 從資料源中擷取對應的字段（包括源表定義的 rowtime）
3. 配置設定 watermark（按照源表定義的 watermark 配置設定對應的 watermark）
4. 将一些必要的字段抽取。比如 group by 中的字段。在 hash 時需要使用。

6

第二個算子：

1. 視窗聚合，計算視窗聚合資料
2. 将資料按照第一層 select 中的資料進行計算以及格式化

7

第三個算子：

1. group 聚合合桶計算
2. 将資料按照第二層 select 中的資料進行計算以及格式化
3. 将資料 sink 寫出

8

3.（flink 1.12.1）結果：

+I(9,1,32682,32682,32682,1,1631026440000)
-U(9,1,32682,32682,32682,1,1631026440000)
+U(9,2,115351,82669,32682,2,1631026440000)
+I(2,1,76148,76148,76148,1,1631026440000)
+I(8,1,79321,79321,79321,1,1631026440000)
+I(a,1,85792,85792,85792,1,1631026440000)
+I(0,1,12858,12858,12858,1,1631026440000)
+I(5,1,36753,36753,36753,1,1631026440000)
+I(3,1,19218,19218,19218,1,1631026440000)
...
           

複制

4.（flink 1.12.1）原理：

關于 sql 開始運作的機制見上一節詳述。

此處隻介紹相比前一節新增内容。可以看到上述代碼的具體 transformation 如下圖。

9

4.4.GeneratedWatermarkGenerator - flink 1.12.1

按照順序，首先看看 watermark 算子。同 datastream 的自定義 watermark 配置設定政策。

10

watermark 生成的具體代碼

WatermarkGenerator$6

，主要擷取 watermark 的邏輯在

currentWatermark

方法中。如下圖。

11

4.5.BinaryRowDataKeySelector - flink 1.12.1

接着就是 group by（同 datastream 中的 keyby）。

12

group by key 生成的具體代碼

KeyProjection$19

，主要邏輯在

apply

方法中。

13

下一個就是視窗聚合算子。

4.6.AggregateWindowOperator - flink 1.12.1

兄弟們！！！兄弟們！！！兄弟們！！！

本節的重頭戲來了。sql 視窗聚合算子解析搞起來了。

關于

WatermarkGenerator

和

KeyProjection

沒有什麼可以詳細介紹的，都是輸入一條資料，輸出一條資料，邏輯很簡單。

但是視窗聚合算子的計算邏輯相比上面兩個算子複雜很多。視窗算子又承載了視窗聚合的主要邏輯，是以本文重點介紹視窗算子計算的邏輯。

先來看看 sql 視窗整體處理流程。其實與 datastream 處理流程基本一緻，但隻是少了

Evictor

。如下圖所示。

40

接着來看看上述 sql 生成的視窗聚合算子

AggregateWindowOperator

，截圖中屬性也很清晰。

16

14

具體生成的視窗聚合代碼

GroupingWindowAggsHandler$59

。

41

計算邏輯

GroupingWindowAggsHandler$59#accumulate

。

42

43

上面那段都是在 flink 用戶端初始化處理的。包括視窗算子的初始化等。

下面這段處理邏輯是在 flink TM 運作時開始執行的，包括視窗算子資源的初始化以及運作邏輯。就到了正式的資料處理環節了。

視窗算子 Task 運作。

27

視窗算子 Task 初始化。

28

StreamTask 整體的處理流程。

29

視窗算子 open 初始化。

30

31

視窗算子 open 初始化後的結果。如下圖，對應的具體元件。

32

初始化完成之後，開始處理具體資料。

循環 loop，一直 run 啊 run。

35

判斷記錄的具體類型，然後執行不同的邏輯。

36

來看看處理一條資料的

processElement

方法邏輯，進行 acc 處理。代碼中的的

windowAggregator

就是之前代碼生成的

GroupingWindowAggsHandler$59

。

Notes：事件時間邏輯中，sql api 和 datastream api 對于資料記錄時間戳存儲邏輯是不一樣的。datastream api：每條記錄的 rowtime 是放在

StreamRecord

中的時間戳字段中的。sql api：時間戳是每次都從資料中進行擷取的。算子中會維護一個下标。可以按照下标從資料中擷取時間戳。

39

來看看 watermark 到達并且觸發視窗計算時，執行

onEventTime

邏輯。

38

觸發視窗計算時，

onEventTime -> emitWindowResult

，産出具體資料。

17

至此整個 sql tumble window 的處理邏輯也就很清楚了。和 datastream 基本上都是一緻的。是不是整個邏輯就理清楚了。

5.總結與展望篇

源碼公衆号背景回複flink sql tumble window 的奇妙解析之路擷取。

本文主要介紹了 tumble window 聚合類名額的常見場景案例以及其底層運作原理。

而且也介紹了在檢視 flink sql 任務時的一些技巧：

1. 去 flink webui 就能看到這個任務目前在做什麼。包括算子名稱都會給直接展示給我們目前哪個算子在幹啥事情，在處理啥邏輯。
2. sql 的 watermark 類型要設定為 TIMESTAMP(3)。
3. 事件時間邏輯中，sql api 和 datastream api 對于資料記錄時間戳存儲邏輯是不一樣的。datastream api：每條記錄的 rowtime 是放在 StreamRecord 中的時間戳字段中的。sql api：時間戳是每次都從資料中進行擷取的。算子中會維護一個下标。可以按照下标從資料中擷取時間戳。

後續文章會基于一些最常見的案例以及原理層面介紹 1.13 版本的 flink sql tumble window（基于最新的 window tvf）。