SQL 動(dòng)態(tài)表 & 連續(xù)查詢

hi，大家好，我是老羊，今天給大家?guī)?lái)一篇關(guān)于 Flink SQL 流式計(jì)算的核心思想設(shè)計(jì)文章。

在小伙伴萌看下文之前，先看一下本文整體的思路，跟著博主思路走，會(huì)更清晰：

• 先分析一下將 SQL 應(yīng)用到流處理的思路
•  SQL 應(yīng)用于批處理已經(jīng)很成熟了，通過(guò)對(duì)比流批處理在輸入、數(shù)據(jù)處理、輸出的異同點(diǎn)來(lái)分析出將 SQL 應(yīng)用于流處理的核心要解決的問(wèn)題點(diǎn)
• 分析如何使用 SQL 動(dòng)態(tài)輸入表 技術(shù)來(lái)將 輸入數(shù)據(jù)流 映射到 SQL 中的輸入表
• 分析如何使用 SQL 連續(xù)查詢 技術(shù)來(lái)將 計(jì)算邏輯 映射到 SQL 中的運(yùn)算語(yǔ)義
• 使用 SQL 動(dòng)態(tài)表 & 連續(xù)查詢技術(shù) 兩種技術(shù)方案來(lái)將 流式 SQL 實(shí)際應(yīng)用到兩個(gè)常見案例中
• 分析 SQL 連續(xù)查詢 的兩種類型：更新(Update)查詢 & 追加(Append)查詢
• 分析如何使用 SQL 動(dòng)態(tài)輸出表 技術(shù)來(lái)將 輸出數(shù)據(jù)流 映射到 SQL 中的輸出表

博主認(rèn)為讀完本節(jié)你應(yīng)該掌握：

• SQL 動(dòng)態(tài)輸入表、SQL 動(dòng)態(tài)輸出表
• SQL 連續(xù)查詢 的兩種類型分別對(duì)應(yīng)的查詢場(chǎng)景及 SQL 語(yǔ)義

1.SQL 應(yīng)用于流處理的思路

在流式 SQL 誕生之前，所有的基于 SQL 的數(shù)據(jù)查詢都是基于批數(shù)據(jù)的，沒有將 SQL 應(yīng)用到流數(shù)據(jù)處理這一說(shuō)法。

那么如果我們想將 SQL 應(yīng)用到流處理中，必然要站在巨人的肩膀(批數(shù)據(jù)處理的流程)上面進(jìn)行，那么具體的分析思路如下：

• 步驟一：先比較 批處理 與 流處理 的異同之處：如果有相同的部分，那么可以直接復(fù)用;不同之處才是我們需要重點(diǎn)克服和關(guān)注的。
• 步驟二：摘出 1 中說(shuō)到的不同之處，分析如果要滿足這個(gè)不同之處，目前有哪些技術(shù)是類似的
• 步驟三：再?gòu)倪@些類似的技術(shù)上進(jìn)一步發(fā)展，以滿足將 SQL 應(yīng)用于流任務(wù)中

博主下文就會(huì)根據(jù)上述三個(gè)步驟來(lái)一步一步介紹 動(dòng)態(tài)表 誕生的背景以及這個(gè)概念是如何誕生的。

2.流批處理的異同點(diǎn)及將 SQL 應(yīng)用于流處理核心解決的問(wèn)題

首先對(duì)比一下常見的 批處理 和 流處理 中 數(shù)據(jù)源(輸入表)、處理邏輯、數(shù)據(jù)匯(結(jié)果表) 的異同點(diǎn)。

對(duì)比上述流批處理之后，我們得到了要將 SQL 應(yīng)用于流式任務(wù)的三個(gè)要解決的核心點(diǎn)：

• SQL 輸入表：分析如何將一個(gè)實(shí)時(shí)的，源源不斷的輸入流數(shù)據(jù)表示為 SQL 中的輸入表。
• SQL 處理計(jì)算：分析將 SQL 查詢邏輯翻譯成什么樣的底層處理技術(shù)才能夠?qū)崟r(shí)的處理流式輸入數(shù)據(jù)，然后產(chǎn)出流式輸出數(shù)據(jù)。
• SQL 輸出表：分析如何將 SQL 查詢輸出的源源不斷的流數(shù)據(jù)表示為一個(gè) SQL 中的輸出表。

將上面 3 個(gè)點(diǎn)總結(jié)一下，也就引出了本節(jié)的 動(dòng)態(tài)表 和 連續(xù)查詢 兩種技術(shù)方案：

• 動(dòng)態(tài)表：源源不斷的輸入、輸出流數(shù)據(jù)映射到 動(dòng)態(tài)表
• 連續(xù)查詢：實(shí)時(shí)處理輸入數(shù)據(jù)，產(chǎn)出輸出數(shù)據(jù)的實(shí)時(shí)處理技術(shù)

3.SQL 流處理的輸入：輸入流映射為 SQL 動(dòng)態(tài)輸入表

動(dòng)態(tài)表。這里的動(dòng)態(tài)其實(shí)是相比于批處理的靜態(tài)(有界)來(lái)說(shuō)的。

• 靜態(tài)表：應(yīng)用于批處理數(shù)據(jù)中，靜態(tài)表可以理解為是不隨著時(shí)間實(shí)時(shí)進(jìn)行變化的。一般都是一天、一小時(shí)的粒度新生成一個(gè)分區(qū)。
• 動(dòng)態(tài)表：動(dòng)態(tài)表是隨時(shí)間實(shí)時(shí)進(jìn)行變化的。是將 SQL 體系中表的概念應(yīng)用到 Flink 上面的的核心點(diǎn)。

來(lái)看一個(gè)具體的案例，下圖顯示了點(diǎn)擊事件流(左側(cè))如何轉(zhuǎn)換為動(dòng)態(tài)表(右側(cè))。當(dāng)數(shù)據(jù)源生成更多的點(diǎn)擊事件記錄時(shí)，映射出來(lái)的動(dòng)態(tài)表也會(huì)不斷增長(zhǎng)，這就是動(dòng)態(tài)表的概念：

Dynamic Table

4.SQL 流處理的計(jì)算：實(shí)時(shí)處理底層技術(shù) - SQL 連續(xù)查詢

連續(xù)查詢。

部分高級(jí)關(guān)系數(shù)據(jù)庫(kù)系統(tǒng)提供了一個(gè)稱為物化視圖(Materialized Views) 的特性。

物化視圖其實(shí)就是一條 SQL 查詢，就像常規(guī)的虛擬視圖 VIEW 一樣。但與虛擬視圖不同的是，物化視圖會(huì)緩存查詢的結(jié)果，因此在請(qǐng)求訪問(wèn)視圖時(shí)不需要對(duì)查詢進(jìn)行重新計(jì)算，可以直接獲取物化視圖的結(jié)果，小伙伴萌可以認(rèn)為物化視圖其實(shí)就是把結(jié)果緩存了下來(lái)。

舉個(gè)例子：批處理中，如果以 Hive 天級(jí)別的物化視圖來(lái)說(shuō)，其實(shí)就是每天等數(shù)據(jù)源 ready 之后，調(diào)度物化視圖的 SQL 執(zhí)行然后產(chǎn)生新的結(jié)果提供服務(wù)。那么就可以認(rèn)為一條表示了輸入、處理、輸出的 SQL 就是一個(gè)構(gòu)建物化視圖的過(guò)程。

映射到我們的流任務(wù)中，輸入、處理邏輯、輸出這一套流程也是一個(gè)物化視圖的概念。相比批處理來(lái)說(shuō)，流處理中，我們的數(shù)據(jù)源表的數(shù)據(jù)是源源不斷的。那么從輸入、處理、輸出的整個(gè)物化視圖的維護(hù)流程也必須是實(shí)時(shí)的。

因此我們就需要引入一種實(shí)時(shí)視圖維護(hù)(Eager View Maintenance)的技術(shù)去做到：一旦更新了物化視圖的數(shù)據(jù)源表就立即更新視圖的結(jié)果，從而保證輸出的結(jié)果也是最新的。

這種 實(shí)時(shí)視圖維護(hù)(Eager View Maintenance)的技術(shù)就叫做 連續(xù)查詢。

注意：

• 連續(xù)查詢(Continuous Query) 不斷的消費(fèi)動(dòng)態(tài)輸入表的的數(shù)據(jù)，不斷的更新動(dòng)態(tài)結(jié)果表的數(shù)據(jù)。
• 連續(xù)查詢(Continuous Query) 的產(chǎn)出的結(jié)果 = 批處理模式在輸入表的上執(zhí)行的相同查詢的結(jié)果。相同的 SQL，對(duì)應(yīng)于同一個(gè)輸入數(shù)據(jù)，雖然執(zhí)行方式不同，但是流處理和批處理的結(jié)果是永遠(yuǎn)都會(huì)相同的。

5.SQL 流處理實(shí)際應(yīng)用：動(dòng)態(tài)表 & 連續(xù)查詢技術(shù)的兩個(gè)實(shí)戰(zhàn)案

例總結(jié)前兩節(jié)，動(dòng)態(tài)表 & 連續(xù)查詢 兩項(xiàng)技術(shù)在一條流 SQL 中的執(zhí)行流程總共包含了三個(gè)步驟，如下圖及總結(jié)所示：

Query

• 第一步：將數(shù)據(jù)輸入流轉(zhuǎn)換為 SQL 中的動(dòng)態(tài)輸入表。這里的轉(zhuǎn)化其實(shí)就是指將輸入流映射(綁定)為一個(gè)動(dòng)態(tài)輸入表。上圖雖然分開畫了，但是可以理解為一個(gè)東西。
• 第二步：在動(dòng)態(tài)輸入表上執(zhí)行一個(gè)連續(xù)查詢，然后生成一個(gè)新的動(dòng)態(tài)結(jié)果表。
• 第三步：生成的動(dòng)態(tài)結(jié)果表被轉(zhuǎn)換回?cái)?shù)據(jù)輸出流。

我們實(shí)際介紹一個(gè)案例來(lái)看看其運(yùn)行方式，以上文介紹到的點(diǎn)擊事件流為例，點(diǎn)擊事件流數(shù)據(jù)的字段如下：

[
  user:  VARCHAR,   // 用戶名
  cTime: TIMESTAMP, // 訪問(wèn) URL 的時(shí)間
  url:   VARCHAR    // 用戶訪問(wèn)的 URL
]

• 第一步，將輸入數(shù)據(jù)流映射為一個(gè)動(dòng)態(tài)輸入表。以下圖為例，我們將點(diǎn)擊事件流(圖左)轉(zhuǎn)換為動(dòng)態(tài)表 (圖右)。當(dāng)點(diǎn)擊數(shù)據(jù)源源不斷的來(lái)到時(shí)，動(dòng)態(tài)表的數(shù)據(jù)也會(huì)不斷的增加。

Dynamic Table

• 第二步，在點(diǎn)擊事件流映射的動(dòng)態(tài)輸入表上執(zhí)行一個(gè)連續(xù)查詢(Continuous Query)，并生成一個(gè)新的動(dòng)態(tài)輸出表。

下面介紹兩個(gè)查詢的案例：

第一個(gè)查詢：一個(gè)簡(jiǎn)單的 GROUP-BY COUNT 聚合查詢，寫過(guò) SQL 的都不會(huì)陌生吧，這種應(yīng)該都是最基礎(chǔ)，最常用的對(duì)數(shù)據(jù)按照類別分組的方法。

如下圖所示 group by 聚合的常用案例。

time

那么本案例中呢，是基于 clicks 表中 user 字段對(duì) clicks 表(點(diǎn)擊事件流)進(jìn)行分組，來(lái)統(tǒng)計(jì)每一個(gè) user 的訪問(wèn)的 URL 的數(shù)量。下面的圖展示了當(dāng) clicks 輸入表來(lái)了新數(shù)據(jù)(即表更新時(shí))，連續(xù)查詢(Continuous Query) 的計(jì)算邏輯。

group agg

當(dāng)查詢開始，clicks 表(左側(cè))是空的。

• 當(dāng)?shù)谝恍袛?shù)據(jù)被插入到 clicks 表時(shí)，連續(xù)查詢(Continuous Query)開始計(jì)算結(jié)果數(shù)據(jù)。數(shù)據(jù)源表第一行數(shù)據(jù) [Mary,./home] 輸入后，會(huì)計(jì)算結(jié)果 [Mary, 1] 插入(insert)結(jié)果表。
• 當(dāng)?shù)诙?[Bob, ./cart] 插入到 clicks 表時(shí)，連續(xù)查詢(Continuous Query)會(huì)計(jì)算結(jié)果 [Bob, 1]，并插入(insert)到結(jié)果表。
• 第三行 [Mary, ./prod?id=1] 輸出時(shí)，會(huì)計(jì)算出[Mary, 2](user 為 Mary 的數(shù)據(jù)總共來(lái)過(guò)兩條，所以為 2)，并更新(update)結(jié)果表，[Mary, 1] 更新成 [Mary, 2]。
• 最后，當(dāng)?shù)谒男袛?shù)據(jù)加入 clicks 表時(shí)，查詢將第三行 [Liz, 1] 插入(insert)結(jié)果表中。

注意上述特殊標(biāo)記出來(lái)的字體，可以看到連續(xù)查詢對(duì)于結(jié)果的數(shù)據(jù)輸出方式有兩種：

• 插入(insert)結(jié)果表
• 更新(update)結(jié)果表

大家對(duì)于 插入(insert)結(jié)果表 這件事都比較好理解，因?yàn)殡x線數(shù)據(jù)都只有插入這個(gè)概念。

但是 更新(update)結(jié)果表 就是離線處理中沒有概念了。這就是連續(xù)查詢中中比較重要一個(gè)概念。后文會(huì)介紹。

接下來(lái)介紹第二條查詢語(yǔ)句。

第二條查詢與第一條類似，但是 group by 中除了 user 字段之外，還 group by 了 tumble，其代表開了個(gè)滾動(dòng)窗口(后面會(huì)詳細(xì)說(shuō)明滾動(dòng)窗口的作用)，然后計(jì)算 url 數(shù)量。

group by user，是按照類別(橫向)給數(shù)據(jù)分組，group by tumble 滾動(dòng)窗口是按時(shí)間粒度(縱向)給數(shù)據(jù)進(jìn)行分組。如下圖所示。

time

圖形化一解釋就很好理解了，兩種都是對(duì)數(shù)據(jù)進(jìn)行分組，一個(gè)是按照 類別 分組，另一種是按照 時(shí)間 分組。

與前面一樣，左邊顯示了輸入表 clicks。查詢每小時(shí)持續(xù)計(jì)算結(jié)果并更新結(jié)果表。clicks 表有三列，user，cTime，url。其中 cTime 代表數(shù)據(jù)的時(shí)間戳，用于給數(shù)據(jù)按照時(shí)間粒度分組。

tumble window

我們的滾動(dòng)窗口的步長(zhǎng)為 1 小時(shí)，即時(shí)間粒度上面的分組為 1 小時(shí)。其中時(shí)間戳在 12:00:00 - 12:59:59 之間有四條數(shù)據(jù)。13:00:00 - 13:59:59 有三條數(shù)據(jù)。14:00:00 - 14:59:59 之間有四條數(shù)據(jù)。

• 當(dāng) 12:00:00 - 12:59:59 數(shù)據(jù)輸入之后，1 小時(shí)的窗口，連續(xù)查詢(Continuous Query)計(jì)算的結(jié)果如右圖所示，將 [Mary, 3]，[Bob, 1] 插入(insert)結(jié)果表。
• 當(dāng) 13:00:00 - 13:59:59 數(shù)據(jù)輸入之后，1 小時(shí)的窗口，連續(xù)查詢(Continuous Query)計(jì)算的結(jié)果如右圖所示，將 [Bob, 1]，[Liz, 2] 插入(insert)結(jié)果表。
• 當(dāng) 14:00:00 - 14:59:59 數(shù)據(jù)輸入之后，1 小時(shí)的窗口，連續(xù)查詢(Continuous Query)計(jì)算的結(jié)果如右圖所示，將 [Mary, 1]，[Bob, 2]，[Liz, 1] 插入(insert)結(jié)果表。

而這個(gè)查詢只有 插入(insert)結(jié)果表 這個(gè)行為。

6.SQL 連續(xù)查詢的兩種類型：更新(Update)查詢 & 追加(Append)查詢

雖然前一節(jié)的兩個(gè)查詢看起來(lái)非常相似(都計(jì)算分組進(jìn)行計(jì)數(shù)聚合)，但它們?cè)谝粋€(gè)重要方面不同：

• 第一個(gè)查詢(group by user)，即(Update)查詢：會(huì)更新先前輸出的結(jié)果，即結(jié)果表流數(shù)據(jù)中包含 INSERT 和 UPDATE 數(shù)據(jù)。小伙伴萌可以理解為 group by user 這條語(yǔ)句當(dāng)中，輸入源的數(shù)據(jù)是一直有的，源源不斷的，同一個(gè) user 的數(shù)據(jù)之后可能還是會(huì)有的，因此可以認(rèn)為此 SQL 的每次的輸出結(jié)果都是一個(gè)中間結(jié)果， 當(dāng)同一個(gè) user 下一條數(shù)據(jù)到來(lái)的時(shí)候，就要用新結(jié)果把上一次的產(chǎn)出中間結(jié)果(舊結(jié)果)給 UPDATE 了。所以這就是 UPDATE 查詢的由來(lái)(其中 INSERT 就是第一條數(shù)據(jù)到來(lái)的時(shí)候，沒有之前的中間結(jié)果，所以是 INSERT)。
• 第二個(gè)查詢(group by user, tumble(xxx))，即(Append)查詢：只追加到結(jié)果表，即結(jié)果表流數(shù)據(jù)中只包含 INSERT 的數(shù)據(jù)。小伙伴萌可以理解為雖然 group by user, tumble(xxx) 上游也是一個(gè)源源不斷的數(shù)據(jù)，但是這個(gè)查詢本質(zhì)上是對(duì)時(shí)間上的劃分，而時(shí)間都是越變?cè)酱蟮?，?dāng)前這個(gè)滾動(dòng)窗口結(jié)束之后，后面來(lái)的數(shù)據(jù)的時(shí)間都會(huì)比這個(gè)滾動(dòng)窗口的結(jié)束時(shí)間大，都?xì)w屬于之后的窗口了，當(dāng)前這個(gè)滾動(dòng)窗口的結(jié)果數(shù)據(jù)就不會(huì)再改變了，因此這條查詢只有 INSERT 數(shù)據(jù)，即一個(gè) Append 查詢。

上面是 Flink SQL 連續(xù)查詢處理機(jī)制上面的兩類查詢方式。我們可以發(fā)現(xiàn)連續(xù)查詢的處理機(jī)制不一樣，產(chǎn)出到結(jié)果表中的結(jié)果數(shù)據(jù)也是不一樣的。針對(duì)上面兩種結(jié)果表的更新方式，F(xiàn)link SQL 提出了 changelog 表的概念來(lái)進(jìn)行兼容。

changelog 表這個(gè)概念其實(shí)就和 MySQL binlog 是一樣的。會(huì)包含 INSERT、UPDATE、DELETE 三種數(shù)據(jù)，通過(guò)這三種數(shù)據(jù)的處理來(lái)描述實(shí)時(shí)處理技術(shù)對(duì)于動(dòng)態(tài)表的變更：

• changelog 表：即第一個(gè)查詢的輸出表，輸出結(jié)果數(shù)據(jù)不但會(huì)追加，還會(huì)發(fā)生更新
• changelog insert-only 表：即第二個(gè)查詢的輸出表，輸出結(jié)果數(shù)據(jù)只會(huì)追加，不會(huì)發(fā)生更新

7.SQL 流處理的輸出：動(dòng)態(tài)輸出表轉(zhuǎn)化為輸出數(shù)據(jù)

可以看到我們的標(biāo)題都是隨著一個(gè) SQL 的生命周期的。從 輸入流映射為 SQL 動(dòng)態(tài)輸入表、實(shí)時(shí)處理底層技術(shù) - SQL 連續(xù)查詢 到本小節(jié)的 SQL 動(dòng)態(tài)輸出表轉(zhuǎn)化為輸出數(shù)據(jù)。都是有邏輯關(guān)系的。

我們上面介紹到了 連續(xù)查詢(Continuous Query) 的輸出結(jié)果表是一個(gè) changelog。其可以像普通數(shù)據(jù)庫(kù)表一樣通過(guò) INSERT、UPDATE 和 DELETE 來(lái)不斷修改。

它可能是一個(gè)只有一行、不斷更新 changelog 表，也可能是一個(gè) insert-only 的 changelog 表，沒有 UPDATE 和 DELETE 修改，或者介于兩者之間的其他表。

在將動(dòng)態(tài)表轉(zhuǎn)換為流或?qū)⑵鋵懭胪獠肯到y(tǒng)時(shí)，需要對(duì)這些不同狀態(tài)的數(shù)據(jù)進(jìn)行編碼。Flink 的 Table API 和 SQL API 支持三種方式來(lái)編碼一個(gè)動(dòng)態(tài)表的變化:

• Append-only 流：輸出的結(jié)果只有 INSERT 操作的數(shù)據(jù)。
• Retract 流：

Retract 流包含兩種類型的 message：add messages 和 retract messages 。其將 INSERT 操作編碼為 add message、將 DELETE 操作編碼為 retract message、將 UPDATE 操作編碼為更新先前行的 retract message 和更新(新)行的 add message，從而將動(dòng)態(tài)表轉(zhuǎn)換為 retract 流。

Retract 流寫入到輸出結(jié)果表的數(shù)據(jù)如下圖所示，有 -，+ 兩種，分別 - 代表撤回舊數(shù)據(jù)，+ 代表輸出最新的數(shù)據(jù)。這兩種數(shù)據(jù)最終都會(huì)寫入到輸出的數(shù)據(jù)引擎中。

如果下游還有任務(wù)去消費(fèi)這條流的話，要注意需要正確處理 -，+ 兩種數(shù)據(jù)，防止數(shù)據(jù)計(jì)算重復(fù)或者錯(cuò)誤。

retract

• Upsert 流：

Upsert 流包含兩種類型的 message：upsert messages 和 delete messages。轉(zhuǎn)換為 upsert 流的動(dòng)態(tài)表需要唯一鍵(唯一鍵可以由多個(gè)字段組合而成)。其會(huì)將 INSERT和 UPDATE 操作編碼為 upsert message，將 DELETE 操作編碼為 delete message。

Upsert 流寫入到輸出結(jié)果表的數(shù)據(jù)如下圖所示，每次輸出的結(jié)果都是當(dāng)前每一個(gè) user 的最新結(jié)果數(shù)據(jù)，不會(huì)有 Retract 中的 - 回撤數(shù)據(jù)。

如果下游還有一個(gè)任務(wù)去消費(fèi)這條流的話，消費(fèi)流的算子需要知道唯一鍵(即 user)，以便正確地根據(jù)唯一鍵(user)去拿到每一個(gè) user 當(dāng)前最新的狀態(tài)。其與 retract 流的主要區(qū)別在于 UPDATE 操作是用單個(gè) message 編碼的，因此效率更高。下圖顯示了將動(dòng)態(tài)表轉(zhuǎn)換為 upsert 流的過(guò)程。

upsert

8.補(bǔ)充知識(shí)：SQL 與關(guān)系代數(shù)

小伙伴萌會(huì)問(wèn)到，關(guān)系代數(shù)是啥東西?

其實(shí)關(guān)系代數(shù)就是對(duì)于數(shù)據(jù)集(即表)的一系列的 操作(即查詢語(yǔ)句)。常見關(guān)系代數(shù)有：

Relational Algebra

那么 SQL 和關(guān)系代數(shù)是啥關(guān)系呢?

SQL 就是能夠表示關(guān)系代數(shù)一種面向用戶的接口：即用戶能使用 SQL 表達(dá)關(guān)系代數(shù)的處理邏輯，也就是我們可以用 SQL 去在表(數(shù)據(jù)集)上執(zhí)行我們的業(yè)務(wù)邏輯操作(關(guān)系代數(shù)操作)。