作者：林小鉑（網易遊戲）
确定性（Determinism）是計算機科學中十分重要的特性，确定性的算法保證對于給定相同的輸入總是産生相同的輸出。在分布式實時計算領域，确定性是業界一直難以解決的課題，由此導緻用離線計算修正實時計算結果的 Lambda 架構成為大資料領域過去近十年的主流架構。
而在最近幾年随着 Google The Dataflow Model 的提出，實時計算和離線計算的關系逐漸清晰，在實時計算中提供與離線計算一緻的确定性成為可能。本文将基于流行實時計算引擎 Apache Flink，梳理建構一個确定性的實時應用要滿足什麼條件。
确定性與準确性
比起确定性，準确性（Accuracy）可能是我們接觸更多的近義詞，大多數場景下兩者可以混用，但其實它們稍有不同: 準确的東西一定是确定的，但确定性的東西未必百分百準确。在大資料領域，不少算法可以根據需求調整成本和準确性的平衡，比如 HyperLogLog 去重統計算法給出的結果是有一定誤差的（是以不是準确的），但卻同時是确定性的（重算可以得到相同結果）。
要分區确定性和準确性的緣故是，準确性與具體的業務邏輯緊密耦合難以評估，而确定性則是通用的需求（除去少數場景使用者故意使用非确定性的算法）。當一個 Flink 實時應用提供确定性，意味着它在異常場景的自動重試或者手動重流資料的情況下，都能像離線作業一般産出相同的結果，這将很大程度上提高使用者的信任度。
影響 Flink 應用确定性的因素
投遞語義
常見的投遞語義有 At-Most-Once、At-Least-Once 和 Exactly-Once 三種。嚴格來說隻有 Exactly-Once滿足确定性的要求，但如果整個業務邏輯是幂等的， 基于At-Least-Once 也可以達到結果的确定性。
實時計算的 Exactly-Once 通常指端到端的 Exactly-Once，保證輸出到下遊系統的資料和上遊的資料是一緻的，沒有重複計算或者資料丢失。要達到這點，需要分别實作讀取資料源（Source 端）的 Exactly-Once、計算的 Exactly-Once 和輸出到下遊系統（Sink 端）的 Exactly-Once。
其中前面兩個都比較好保證，因為 Flink 應用出現異常會自動恢複至最近一個成功 checkpoint，Pull-Based 的 Source 的狀态和 Flink 内部計算的狀态都會自動復原到快照時間點，而問題在于 Push-Based 的 Sink 端。Sink 端是否能順利復原依賴于外部系統的特性，通常來說需要外部系統支援事務，然而不少大資料元件對事務的支援并不是很好，即使是實時計算最常用的 Kafka 也直到 2017 年的 0.11 版本才支援事務，更多的元件需要依賴各種 trick 來達到某種場景下的 Exactly-Once。
總體來說這些 trick 可以分為兩大類:
- 依賴寫操作的幂等性。比如 HBase 等 KV 存儲雖然沒有提供跨行事務，但可以通過幂等寫操作配合基于主鍵的 Upsert 操作達到 Exactly-Once。不過由于 Upsert 不能表達 Delete 操作，這種模式不适合有 Delete 的業務場景。
- 預寫日志（WAL，Write-Ahead-Log）。預寫日志是廣泛應用于事物機制的技術，包括 MySQL、PostgreSQL 等成熟關系型資料庫的事物都基于預寫日志。預寫日志的基本原理先将變更寫入緩存區，等事務送出的時候再一次全部應用。比如 HDFS/S3 等檔案系統本身并不提供事務，是以實作預寫日志的重擔落到了它們的使用者（比如 Flink）身上。通過先寫臨時的檔案/對象，等 Flink Checkpoint 成功後再送出，Flink 的 FileSystem Connector 實作了 Exactly-Once。然而，預寫日志隻能保證事務的原子性和持久性，不能保證一緻性和隔離性。為此 FileSystem Connector 通過将預寫日志設為隐藏檔案的方式提供了隔離性，至于一緻性（比如臨時檔案的清理）則無法保證。
為了保證 Flink 應用的确定性，在選用官方 Connector，特别是 Sink Connector 時，使用者應該留意官方文檔關于 Connector 投遞語義的說明[3]。此外，在實作定制化的 Sink Connector 時也需要明确達到何種投遞語義，可以參考利用外部系統的事務、寫操作的幂等性或預寫日志三種方式達到 Exactly-Once 語義。
函數副作用
函數副作用是指使用者函數對外界造成了計算架構意料之外的影響。比如典型的是在一個 Map 函數裡将中間結果寫到資料庫，如果 Flink 作業異常自動重新開機，那麼資料可能被寫兩遍，導緻不确定性。對于這種情況，Flink 提供了基于 Checkpoint 的兩階段送出的鈎子（CheckpointedFunction 和 CheckpointListener），使用者可以用它來實作事務，以消除副作用的不确定性。另外還有一種常見的情況是，使用者使用本地檔案來儲存臨時資料，這些資料在 Task 重新排程的時候很可能丢失。其他的場景或許還有很多，總而言之，如果需要在使用者函數裡改變外部系統的狀态，請確定 Flink 對這些操作是知情的（比如用 State API 記錄狀态，設定 Checkpoint 鈎子）。
Processing Time
在算法中引入目前時間作為參數是常見的操作，但在實時計算中引入目前系統時間，即 Processing Time，是造成不确定性的最常見也最難避免的原因。對 Processing 的引用可以是很明顯、有完善文檔标注的，比如 Flink 的 Time Characteristic，但也可能是完全出乎使用者意料的，比如來源于緩存等常用的技術。為此，筆者總結了幾類常見的 Processing Time 引用:
- Flink 提供的 Time Characteristic。Time Characteristic 會影響所有使用與時間相關的算子，比如 Processing Time 會讓視窗聚合使用目前系統時間來配置設定視窗和觸發計算，造成不确定性。另外，Processing Timer 也有類似的影響。
- 直接在函數裡通路外部存儲。因為這種通路是基于外部存儲某個 Processing Time 時間點的狀态，這個狀态很可能在下次通路時就發生了變化，導緻不确定性。要獲得确定性的結果，比起簡單查詢外部存儲的某個時間點的狀态，我們應該擷取它狀态變更的曆史，然後根據目前 Event Time 去查詢對應的狀态。這也是 Flink SQL 中 Temporary Table Join 的實作原理[1]。
- 對外部資料的緩存。在計算流量很大的資料時，很多情況下使用者會選擇用緩存來減輕外部存儲的負載，但這可能會造成查詢結果的不一緻，而且這種不一緻是不确定的。無論是使用逾時門檻值、LRU（Least Recently Used）等直接和系統時間相關的緩存剔除政策，還是 FIFO（First In First Out）、LFU（Less Frequently Used）等沒有直接關聯時間的剔除政策，通路緩存得到的結果通常和消息的到達順序相關，而在上遊經過 shuffle 的算子裡面這是難以保證的（沒有 shuffle 的 Embarrassingly Parallel 作業是例外）。
- Flink 的 StateTTL。StateTTL 是 Flink 内置的根據時間自動清理 State 的機制，而這裡的時間目前隻提供 Processing Time，無論 Flink 本身使用的是 Processing Time 還是 Event Time 作為 Time Characteristic。BTW，StateTTL 對 Event Time 的支援可以關注 FLINK-12005[2]。
綜合來講，要完全避免 Processing Time 造成的影響是非常困難的，不過輕微的不确定性對于業務來說通常是可以接受的，我們要做的更多是提前預料到可能的影響，保證不确定性在可控範圍内。
Watermark
Watermark 作為計算 Event Time 的機制，其中一個很重要的用途是決定實時計算何時要輸出計算結果，類似檔案結束标志符（EOF）在離線批計算中達到的效果。然而，在輸出結果之後可能還會有遲到的資料到達，這稱為視窗完整性問題（Window Completeness）。
視窗完整性問題無法避免，應對辦法是要麼更新計算結果，要麼丢棄這部分資料。因為離線場景延遲容忍度較大，離線作業可以推遲一定時間開始，盡可能地将延遲資料納入計算。而實時場景對延遲有比較高的要求，是以一般是輸出結果後讓狀态儲存一段時間，在這段時間内根據遲到資料持續更新結果（即 Allowed Lateness），此後将資料丢棄。因為定位，實時計算天然可能出現更多被丢棄的遲到資料，這将和 Watermark 的生成算法緊密相關。
雖然 Watermark 的生成是流式的，但 Watermark 的下發是斷點式的。Flink 的 Watermark 下發政策有 Periodic 和 Punctuated 兩種，前者基于 Processing Time 定時觸發，後者根據資料流中的特殊消息觸發。
圖1. Periodic Watermark 正常狀态與重放追資料狀态
基于 Processing Time 的 Periodic Watermark 具有不确定。在平時流量平穩的時候 Watermark 的提升可能是階梯式的（見圖1(a)）；然而在重放曆史資料的情況下，相同長度的系統時間内處理的資料量可能會大很多（見圖1(b))，并且伴有 Event Time 傾斜（即有的 Source 的 Event Time 明顯比其他要快或慢，導緻取最小值的總體 Watermark 被慢 Watermark 拖慢），導緻本來丢棄的遲到資料，現在變為 Allowed Lateness 之内的資料（見圖1中紅色元素）。
圖2. Punctuated Watermark 正常狀态與重放追資料狀态
相比之下 Punctuated Watermark 更為穩定，無論在正常情況（見圖2(a)）還是在重放資料的情況（見圖2(b)）下，下發的 Watermark 都是一緻的，不過依然有 Event Time 傾斜的風險。對于這點，Flink 社群起草了 FLIP-27 來處理[4]。基本原理是 Source 節點會選擇性地消費或阻塞某個 partition/shard，讓總體的 Event Time 保持接近。
除了 Watermark 的下發有不确定之外，還有個問題是現在 Watermark 并沒有被納入 Checkpoint 快照中。這意味着在作業從 Checkpoint 恢複之後，Watermark 會重新開始算，導緻 Watermark 的不确定。這個問題在 FLINK-5601[5] 有記錄，但目前隻展現了 Window 算子的 Watermark，而在 StateTTL 支援 Event Time 後，或許每個算子都要記錄自己的 Watermark。
綜上所述，Watermark 目前是很難做到非常确定的，但因為 Watermark 的不确定性是通過丢棄遲到資料導緻計算結果的不确定性的，隻要沒有丢棄遲到資料，無論中間 Watermark 的變化如何，最終的結果都是相同的。
總結
确定性不足是阻礙實時計算在關鍵業務應用的主要因素，不過目前業界已經具備了解決問題的理論基礎，剩下的更多是計算架構後續疊代和工程實踐上的問題。就目前開發 Flink 實時應用而言，需要注意投遞語義、函數副作用、Processing Time 和 Watermark 這幾點造成的不确定性。
參考：
- Flux capacitor, huh? Temporal Tables and Joins in Streaming SQL https://flink.apache.org/2019/05/14/temporal-tables.html
- FLINK-12005 Event time support https://issues.apache.org/jira/browse/FLINK-12005
- Fault Tolerance Guarantees of Data Sources and Sinks https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/connectors/guarantees.html
- FLIP-27: Refactor Source Interface https://cwiki.apache.org/confluence/display/FLINK/FLIP-27%3A+Refactor+Source+Interface
- [FLINK-5601] Window operator does not checkpoint watermarks https://issues.apache.org/jira/browse/FLINK-5601
作者介紹：
林小鉑，網易遊戲進階開發工程師，負責遊戲資料中心實時平台的開發及運維工作，目前專注于 Apache Flink 的開發及應用。探究問題本來就是一種樂趣。