Flink Table 的三種 Sink 模式

本文原創做者：林小鉑(Paul Lin)，現就任於網易遊戲負責遊戲數據中心實時平臺開發維護工做。點擊閱讀原文，能夠連接到做者的我的博客閱讀。

做爲計算引擎 Flink 應用的計算結果總要以某種方式輸出，好比調試階段的打印到控制檯或者生產階段的寫到數據庫。而對於原本就須要在 Flink 內存保存中間及最終計算結果的應用來講，好比進行聚合統計的應用，輸出結果即是將內存中的結果同步到外部。就 Flink Table/SQL API 而言，這裏的同步會有三種模式，分別是 Append、Upsert 和 Retract。實際上這些輸出計算結果的模式並不限於某個計算框架，好比 Storm、Spark 或者 Flink DataStream 均可以應用這些模式，不過 Flink Table/SQL 已有完整的概念和內置實現，更方便討論。

基礎原理

相信接觸過 Streaming SQL 的同窗都有了解或者聽過流表二象性，簡單來講流和表是同一事實的不一樣表現，是能夠相互轉換的。流和表的表述在業界不盡相同，筆者比較喜歡的一種是: 流體現事實在時間維度上的變化，而表則體現事實在某個時間點的視圖。若是將流比做水管中流動的水，那麼表將是杯子裏靜止的水。

將流轉換爲表的方法對於大多數讀者都不陌生，只需將聚合統計函數應用到流上，流很天然就變爲表（值得注意的是，Flink 的 Dynamic Table 和表的定義有細微不一樣，這將在下文講述）。好比對於一個計算 PV 的簡單流計算做業，將用戶瀏覽日誌數據流安 url 分類統計，變成 (url, views) 這樣的一個表。然而對於如何將錶轉換成流，讀者則未必有這麼清晰的概念。

假設一個典型的實時流計算應用的工做流程能夠被簡化爲下圖:

其中很關鍵的一點是 Transformation 是否聚合類型的計算。若否，則輸出結果依然是流，能夠很天然地使用本來流處理的 Sink（與外部系統的鏈接器）；如果，則流會轉換爲表，那麼輸出的結果將是表，而一個表的輸出一般是批處理的概念，不能直接簡單地用流處理的 Sink 來表達。

這時有個很樸素的想法是，咱們能不能避免批處理那種全量的輸出，每次只輸出表的 diff，也就是 changelog。這也是錶轉化爲流的方法: 持續觀察表的變化，並將每一個變化記錄成日誌輸出。所以，流和表的轉換能夠如下圖表示:

其中表的變化具體能夠分爲 INSERT、UPDATE 和 DELETE 三類，而 Flink 根據這些變化類型分別總結了三種結果的輸出模式。

模式	INSERT	UPDATE	DELETE
Append	支持	不支持	不支持
Upsert	支持	支持	支持
Retract	支持	支持	支持

一般來講 Append 是最容易實現但功能最弱的，Retract 是最難實現而功能最強的。下文分別談談三種模式的特色和應用場景。

Append 輸出模式

Append 是最爲簡單的輸出模式，只支持追加結果記錄的操做。由於結果一旦輸出之後便不會再有變動，Append 輸出模式的最大特性是不可變性（immutability），而不可變性最使人嚮往的優點即是安全，好比線程安全或者 Event Sourcing 的可恢復性，不過同時也會給業務操做帶來限制。一般來講，Append 模式會用於寫入不方便作撤回或者刪除操做的存儲系統的場景，好比 Kafka 等 MQ 或者打印到控制檯。

在實時聚合統計中，聚合統計的結果輸出是由 Trigger 決定的，而 Append-Only 則意味着對於每一個窗口實例（Pane，窗格）Trigger 只能觸發一次，則就致使沒法在遲到數據到達時再刷新結果。一般來講，咱們能夠給 Watermark 設置一個較大的延遲容忍閾值來避免這種刷新（再有遲到數據則丟棄），但代價是卻會引入較大的延遲。

不過對於不涉及聚合的 Table 來講，Append 輸出模式是很是好用的，由於這類 Table 只是將數據流的記錄按時間順序排在一塊兒，每條記錄間的計算都是獨立的。值得注意的是，從 DataFlow Model 的角度來看未作聚合操做的流不該當稱爲表，可是在 Flink 的概念裏全部的流均可以稱爲 Dynamic Table。筆者認爲這個設計也有必定的道理，緣由是從流中截取一段出來依然能夠知足表的定義，即」某個時間點的視圖」，並且咱們能夠爭辯說不聚合也是一種聚合函數。

Upsert 輸出模式

Upsert 是 Append 模式的升級版，支持 Append-Only 的操做和在有主鍵的前提下的 UPDATE 和 DELETE 操做。Upsert 模式依賴業務主鍵來實現輸出結果的更新和刪除，所以很是適合 KV 數據庫，好比
HBase、JDBC 的 TableSink 都使用了這種方式。

在底層，Upsert 模式下的結果更新會被翻譯爲 (Boolean, ROW) 的二元組。其中第一個元素表示操做類型，true 對應 UPSERT 操做（不存在該元素則 INSERT，存在則 UPDATE），false 對應 DELETE 操做，第二個元素則是操做對應的記錄。若是結果表自己是 Append-Only 的，第一個元素會所有爲 true，並且也無需提供業務主鍵。

Upsert 模式是目前來講比較實用的模式，由於大部分業務都會提供原子或複合類型的主鍵，而在支持 KV 的存儲系統也很是多，但要注意的是不要變動主鍵，具體緣由會在下一節談到。

Retract 輸出模式

Retract 是三種輸出模式中功能最強大但實現也最複雜的一種，它要求目標存儲系統能夠追蹤每一個條記錄，並且這些記錄至少在必定時間內都是能夠撤回的，所以一般來講它會自帶系統主鍵，沒必要依賴於業務主鍵。然而因爲大數據存儲系統不多有能夠精確到一條記錄的更新操做，所以目前來講至少在 Flink 原生的 TableSink 中尚未能在生產環境中知足這個要求的。

不一樣於 Upsert 模式更新時會將整條記錄從新輸出，Retract 模式會將更新分紅兩條表示增減量的消息，一條是 (false, OldRow) 的撤回（Retract）操做，一條是 (true, NewRow) 的積累（Accumulate）操做。這樣的好處是，在主鍵出現變化的狀況下，Upsert 輸出模式沒法撤回舊主鍵的記錄，致使數據不許確，而 Retract 模式則不存在這個問題。

舉個例子，假設咱們將電商訂單按照承運快遞公司進行分類計數，有以下的結果表。

那麼若是本來一單爲中通的快遞，後續更新爲用順豐發貨，對於 Upsert 模式會產生 (true, (順豐, 4)) 這樣一條 changelog，但中通的訂單數沒有被修正。相比之下，Retract 模式產出 (false, (中通, 1)) 和 (true, (順豐, 1)) 兩條數據，則能夠正確地更新數據。

總結

Flink Table Sink 的三種模式本質上是如何監控結果表併產生 changelog，這能夠應用於全部須要將錶轉爲流的場景，包括同一個 Flink 應用的不一樣表間的聯動。三種模式中 Append 模式只支持表的 INSERT，最爲簡單；Upsert 模式依賴業務主鍵提供 INSERT、UPDATE 和 DELETE 所有三類變動，比較實用；Retract 模式一樣支持三類變動且不要求業務主鍵，但會將 UPDATE 翻譯爲舊數據的撤回和新數據的累加，實現上比較複雜。

參考

1. Stream Processing, Event Sourcing, and Data Streaming Explained(https://www.confluent.io/blog/making-sense-of-stream-processing/)

2. Introducing Kafka Streams: Stream Processing Made Simple(https://www.confluent.io/blog/introducing-kafka-streams-stream-processing-made-simple/)

3. Dynamic Tables(https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/streaming/dynamic_tables.html)

4. Flink 源碼閱讀筆記（18）- Flink SQL 中的流和動態表(https://blog.jrwang.me/2019/2019-10-16-flink-sourcecode-stream-and-dynamic-table/)

         

本文分享自微信公衆號 - ApacheHudi（ApacheHudi）。
若有侵權，請聯繫 support@oschina.cn 刪除。
本文參與「OSC源創計劃」，歡迎正在閱讀的你也加入，一塊兒分享。