Flink/Spark 如何實現動態更新作業配置
由於實時場景對可用性十分敏感，實時作業通常需要避免頻繁重啟，因此動態加載作業配置（變量）是實時計算里十分常見的需求，比如通常復雜事件處理 (CEP) 的規則或者在線機器學習的模型。盡管常見，實現起來卻並沒有那么簡單，其中最難點在於如何確保節點狀態在變更期間的一致性。目前來說一般有兩種實現方式：
- 輪詢拉取方式，即作業算子定時檢測在外部系統的配置是否有變更，若有則同步配置。
- 控制流方式，即作業除了用於計算的一個或多個普通數據流以外，還有提供一個用於改變作業算子狀態的元數據流，也就是控制流。
輪詢拉取方式基於 pull 模式，一般實現是用戶在 Stateful 算子(比如 RichMap)里實現后台線程定時從外部系統同步變量。這種方式對於一般作業或許足夠，但存在兩個缺點分別限制了作業的實時性和准確性的進一步提高：首先，輪詢總是有一定的延遲，因此變量的變更不能第一時間生效；其次，這種方式依賴於節點本地時間來進行校准。如果在同一時間有的節點已經檢測到變更並更新狀態，而有的節點還沒有檢測到或者還未更新，就會造成短時間內的不一致。
控制流方式基於 push 模式，變更的檢測和節點更新的一致性都由計算框架負責，從用戶視角看只需要定義如何更新算子狀態並負責將控制事件丟入控制流，后續工作計算框架會自動處理。控制流不同於其他普通數據流的地方在於控制流是以廣播形式流動的，否則在有 Keyby 或者 rebalance 等提高並行度分流的算子的情況下就無法將控制事件傳達給所有的算子。
以目前最流行的兩個實時計算框架 Spark Streaming 和 Flink 來說，前者是以類似輪詢的方式來實現實時作業的更新，而后者則是基於控制流的方式。
Spark Streaming Broadcast Variable
Spark Streaming 為用戶提供了 Broadcast Varialbe，可以用於節點算子狀態的初始化和后續更新。Broacast Variable 是一組只讀的變量，它在作業初始化時由 Spark Driver 生成並廣播到每個 Executor 節點，隨后該節點的 Task 可以復用同一份變量。
Broadcast Variable 的設計初衷是為了避免大文件，比如 NLP 常用的分詞詞典，隨序列化后的作業對象一起分發，造成重復分發的網絡資源浪費和啟動時間延長。這類文件的更新頻率是相對低的，扮演的角色類似於只讀緩存，通過設置 TTL 來定時更新，緩存過期之后 Executor 節點會重新向 Driver 請求最新的變量。
Broadcast Variable 並不是從設計理念上就支持低延遲的作業狀態更新，因此用戶想出了不少 Hack 的方法，其中最為常見的方式是：一方面在 Driver 實現后台線程不斷更新 Broadcast Variavle，另一方面在作業運行時通過顯式地刪除 Broadcast Variable 來迫使 Executor 重新從 Driver 拉取最新的 Broadcast Variable。這個過程會發生在兩個 micro batch 計算之間，以確保每個 micro batch 計算過程中狀態是一致的。
比起用戶在算子內訪問外部系統實現更新變量，這種方式的優點在於一致性更有保證。因為 Broadcast Variable 是統一由 Driver 更新並推到 Executor 的，這就保證不同節點的更新時間是一致的。然而相對地，缺點是會給 Driver 帶來比較大的負擔，因為需要不斷分發全量的 Broadcast Variable (試想下一個巨大的 Map，每次只會更新少數 Entry，卻要整個 Map 重新分發)。在 Spark 2.0 版本以后，Broadcast Variable 的分發已經從 Driver 單點改為基於 BitTorrent 的 P2P 分發，這一定程度上緩解了隨着集群規模提升 Driver 分發變量的壓力，但我個人對這種方式能支持到多大規模的部署還是持懷疑態度。另外一點是重新分發 Broadcast Variable 需要阻塞作業進行，這也會使作業的吞吐量和延遲受到比較大的影響。
Flink Broadcast State & Stream
Broadcast Stream 是 Flink 1.5.0 發布的新特性，基於控制流的方式實現了實時作業的狀態更新。Broadcast Stream 的創建方式與普通數據流相同，例如從 Kafka Topic 讀取，特別之處在於它承載的是控制事件流，會以廣播形式將數據發給下游算子的每個實例。Broadcast Stream 需要在作業拓撲的某個節點和普通數據流 (Main Stream) join 到一起。
該節點的算子需要同時處理普通數據流和控制流：一方面它需要讀取控制流以更新本地狀態 (Broadcast State)，另外一方面需要讀取 Main Stream 並根據 Broadcast State 來進行數據轉換。由於每個算子實例讀到的控制流都是相同的，它們生成的 Broadcast State 也是相同的，從而達到通過控制消息來更新所有算子實例的效果。
目前 Flink 的 Broadcast Stream 從效果上實現了控制流的作業狀態更新，不過在編程模型上有點和一般直覺不同。原因主要在於 Flink 對控制流的處理方式和普通數據流保持了一致，最為明顯的一點是控制流除了改變本地 State 還可以產生 output，這很大程度上影響了 Broadcast Stream 的使用方式。Broadcast Stream 的使用方式與普通的 DataStream 差別比較大，即需要和 DataStream 連接成為 BroadcastConnectedStream 后，再通過特殊的 BroadcastProcessFunction 來處理，而 BroadcastProcessFunction 目前只支持 類似於 RichCoFlatMap 效果的操作。RichCoFlatMap 可以間接實現對 Main Stream 的 Map 轉換（返回一只有一個元素的集合）和 Filter 轉換（返回空集合），但無法實現 Window 類計算。這意味着如果用戶希望改變 Window 算子的狀態，那么需要將狀態管理提前到上游的 BroadcastProcessFunction，然后再通過 BroadcastProcessFunction 的輸出來將影響下游 Window 算子的行為。
總結
實時作業運行時動態加載變量可以令大大提升實時作業的靈活性和適應更多應用場景，目前無論是 Flink 還是 Spark Streaming 對動態加載變量的支持都不是特別完美。Spark Streaming 受限於 Micro Batch 的計算模型（雖然現在 2.3 版本引入 Continuous Streaming 來支持流式處理，但離成熟還需要一定時間），將作業變量作為一致性和實時性要求相對低的節點本地緩存，並不支持低延遲地、低成本地更新作業變量。Flink 將變量更新視為特殊的控制事件流，符合 Even Driven 的流式計算框架定位，目前在業界已有比較成熟的應用。不過美中不足的是編程模型的易用性上有提高空間：控制流目前只能用於和數據流的 join，這意味着下游節點無法繼續訪問控制流或者需要把控制流數據插入到數據流中（這種方式並不優雅），從而降低了編程模型的靈活性。個人認為最好的情況是大部分的算子都可以被拓展為具有 BroadcastOperator，就像 RichFunction 一樣，它們可以接收一個數據流和一個至多個控制流，並維護對應的 BroadcastState，這樣控制流的接入成本將顯著下降。
參考文獻
1.FLIP-17 Side Inputs for DataStream API
2.Dynamically Configured Stream Processing: How BetterCloud Built an Alerting System with Apache Flink®
3.Using Control Streams to Manage Apache Flink Applications
4.StackOverFlow - ow can I update a broadcast variable in spark streaming?