美團基于 Flink 的實時數倉平臺建設新進展
?摘要：本文整理自美團實時數倉平臺負責人姚冬陽在 Flink Forward Asia 2021 實時數倉專場的演講。主要內容包括：
- 平臺建設現狀
- 遇到的問題及解決
- 未來規劃
01平臺建設現狀
美團于 2018 年首次引入 Flink 實時計算引擎，當時的實時數倉概念還不太普及，平臺只提供了 Flink Jar 任務的生命周期管理和監控報警。
2019 年，我們注意到實時計算的主要應用場景是解決離線數倉時效性低的問題。離線數倉已經比較成熟，通過 SQL 方式開發很簡單，而數倉的實時部分主要通過 Flink DataStream API 來開發，門檻比較高，而且與離線數倉的開發方式相比較為割裂。因此，我們開始調研實時數倉的解決方案，目標是降低開發門檻，并嘗試推廣 FlinkSQL，最終將美團的實時數倉平臺取名為 NAU。
2020 年，美團實時數倉平臺正式上線。它向業務提供 FlinkSQL 作業開發入口，主要負責兩個方面的工作：
- 首先，將實時數倉常見的數據源與離線表概念對齊，用數據模型進行管理；
- 其次，提供 FlinkSQL 開發配套的效率工具，比如校驗和調試功能。
但是在實際推廣過程中，我們發現業務在 FlinkSQL 的運維方面門檻依然比較高，因此，我們將接下來的工作重點轉向了運維中心。
FlinkSQL 作業運維的痛點主要集中在兩個方面：有狀態 SQL 作業部署的斷流問題和 SQL 作業的異常定位問題。為此，我們通過 Checkpoint 持久化和狀態生成的異步化來解決第一個問題，并通過提供作業的自動診斷來解決第二個問題。目前，整個實時數倉的平臺化建設已經初步完備，未來我們會在開發和運維能力上不斷精細化，并且繼續推動公司業務數倉架構的進化，比如流批生產的一體化、生產服務的一體化。
實時數倉目前已基本覆蓋了公司的全部業務，為 100 多個業務團隊提供了支持，比如美團優選、美團買菜、金融、騎行等業務。托管了 7000 多個實時數據模型，主要為 Kafka 表和 KV 表模型。線上運行 FlinkSQL 作業 4000+，新增的實時 SQL 作業占比已經達到 70% 以上。從數據上看，FlinkSQL 已經可以解決美團實時數倉大部分流處理的問題。
接下來以美團業務中的兩個實時數倉生產鏈路為例，具體分享 FlinkSQL 的實際應用。
應用場景 1 是基于 FlinkSQL + OLAP 的實時生產鏈路。這個業務鏈路的實時數據源有兩個，分別是業務 DB 的變更事件和業務服務的日志事件，這些事件首先會被收集到 Kafka 中，然后 DB 事件會按表名分發到新的 Kafka 中，DB 和日志的數據也會在這一層進行格式上的統一并完成實時數倉的 ODS 層。然后業務會使用 FlinkSQL 來清洗和關聯 ODS 層的數據，生成實時數倉的主題寬表，最后寫入 OLAP 查詢引擎做實時分析。對于時效性要求不高的場景，部分業務還會在 OLAP 引擎上配置分鐘級別的調度來減少相同查詢的壓力。
應用場景 2 與場景 1 的不同點在于，業務實時數倉的主題寬表數據并不是直接寫入 OLAP 查詢引擎，而是繼續寫入 Kafka，使用 FlinkSQL 做 APP 層的指標聚合，最終把預計算的指標數據寫入 OLAP、DB 或 KV 這類應用層的存儲。這種方式更適合對接數據服務，因為它兼顧了數據的時效性和高 QPS 的查詢。
上圖是實時數倉平臺的架構，分為集成、開發、運維、治理、安全 5 個模塊分別建設。
集成模塊主要關注的是數據模型的管理，具體包括 Kafka 和 KV 兩種模型管理，管理的內容有數據源的 schema 信息和連接信息等。
開發模塊主要關注的是 FlinkSQL 轉化業務需求，比如提供版本管理來記錄業務需求的迭代過程，提供 FlinkSQL 的校驗和調試，來確保開發的 SQL 正確表達了業務邏輯，支持業務使用自定義的 Flink UDF 函數和自定義的 Format 解析，讓 FlinkSQL 可以擴展滿足更多業務需求場景。
運維模塊關注的是 SQL 作業的部署和運行時的監控。在監控方面，我們提供了 SQL 作業的監控報警、異常日志和作業診斷，能夠幫助業務快速發現和定位作業的異常；部署方面，我們提供 SQL 作業的快照管理、AB 部署和參數調優，來幫助業務解決 SQL 作業變更時的問題。
治理模塊關注的是實時數倉的數據質量、資源成本，通過建設實時數倉的 DQC 監控，幫助業務發現上游數據或產出數據的異常值/異常波動；通過鏈路血緣和資源計費，讓業務可以量化實時數倉的生產成本，方便進行成本治理。
安全模塊主要關注的是對數據流向的管控，提供數據源讀寫權限的管理和受限域機制，保證公司業務數據的安全性。
02遇到的問題及解決
在實際推廣 FlinkSQL 的過程中，我們也面臨了不少挑戰。
2.1 雙流關聯大狀態問題
首先是雙流關聯的大狀態問題，FlinkSQL 的雙流關聯會保留左右流的歷史數據來互相關聯，需要關聯的時間間隔越長，保存的歷史數據就會越多，狀態也就會越大。比如，要關聯訂單的下單事件和退款事件，并保證計算結果的正確性，需要考慮這兩個事件發生的間隔，可能是一個月甚至更久。
上圖左側是一個雙流關聯的有狀態 SQL 作業，圖中的 Mem 和 Disk 組成了 SQL 作業的 TaskManager 節點，SQL 作業狀態后端使用 RocksDB，狀態持久化在 HDFS 文件系統上。一開始我們嘗試把 SQL 作業的狀態設置為保留一個月，但 SQL 作業會變得不穩定，出現內存超限、狀態讀取性能下降等問題，只能不斷增加作業的 TM 數和內存大小來緩解。
即使這樣，業務上仍然存在兩個痛點。首先是關聯數據初始化難，目前公司 Kafka 數據源對歷史回溯有限制，因此業務不能構建出完整的歷史狀態，即使 Kafka 支持了更久的回溯，狀態初始化的效率也依然是一個問題。其次，內存資源開銷大，特別是當多個 SQL 作業關聯相同的數據源時，需要為每個 SQL 作業都分配相應的內存資源，不同 SQL 作業間的狀態是隔離的，作業間相同的關聯數據不能復用。
解決方案對于上述問題，我們提出了冷熱關聯分離的解決方案。假設關聯兩天前的數據是相對低頻的且狀態回滾不會超過兩天，那么可以定義兩天前的數據為冷數據，兩天之內的數據為熱數據。
如上圖所示，左側的 SQL 作業通過設置狀態保留時長，只保留 T+0 和 T+1 這兩天的熱數據，而 T+2 及更久以前的冷數據則通過批任務每天從 Hive 同步到外存 KV 中。關聯時，若狀態中的熱數據不存在，則再通過訪問外存 KV 來關聯冷數據。右側是另外一個 SQL 作業需要關聯相同的數據源，它與左側的 SQL 作業共享外層 KV 中的冷數據。
對于第一個痛點，因為狀態控制在了兩天內，SQL 作業上線時，關聯數據初始化的數據量得到了控制。對于第二個痛點，因為兩天前的大部分數據都保存在外層KV中，不同的 SQL 作業都可以查詢外存 KV，從而可以節省大量內存資源。
2.2 SQL 變更狀態恢復問題
第二個問題是有狀態 SQL 邏輯變更后狀態如何恢復？FlinkSQL 支持有狀態的增量計算，狀態是增量計算的歷史累計，實際上業務需要修改邏輯的情況很多，上圖右側列出了一些常見的 SQL 變更情況，比如新增聚合指標、修改原指標口徑、增加過濾條件、新增數據流關聯、增加聚合維度等。
舉個例子，業務增加了更多服務維度，在數據產品上就需要擴展分析的維度，因此也需要修改 FlinkSQL 增加聚合維度。但是上述 SQL 邏輯變化后卻不能從之前的狀態恢復，因為歷史狀態對于變更后的 SQL 不能保證其完整性，即使恢復后也不能百分百保證后續計算的正確性。這種情況下，業務為了保證數據的正確性，需要從歷史回溯重新計算，回溯的過程會導致線上斷流，但業務又不希望犧牲太多的時效性。
解決方案針對這個問題，我們給出了三種解決方案。
解法 1：雙鏈路切換。此解法的關鍵是再搭建一條相同的實時鏈路作為備用鏈路，當變更有狀態 SQL 時，可以在備用鏈路上做回溯，重新計算歷史數據，回溯完成后先驗證備用鏈路的結果數據，確保沒問題后再在鏈路最下游的數據服務層切換讀取的表，完成整個變更流程。
解法 2：旁路狀態生成。與雙鏈路切換不同點在于，這里變更的是鏈路上的單個作業，思路是臨時啟動一個旁路作業來回溯，構建出新邏輯的狀態，驗證數據完成后再重啟線上作業，以此完成 SQL 和狀態的同時切換。
解法 3：歷史狀態遷移，前兩個方法的思路比較類似，都是基于歷史數據重新計算，構建出新狀態。但這個思路是基于歷史狀態遷移出新狀態，這種方法構建出的新狀態雖然不能保證完整性，但在某些情況下，業務也是可以接受的。目前我們通過改造 State Process API 支持在 SQL 算子及其上下游關系不變的情況下，允許 Join 和 Agg 算子來新增列。
上述三種方式各有優點，可以從普適性、資源成本、線上斷流、等待時長四個維度來對以上三個解決方案進行橫向比較。
普適性是指在保證數據正確的前提下支持的 SQL 變更范圍，前兩個方法都是重新計算，狀態是完整的，因此比方案 3 的普適性更高。
資源成本是指完成 SQL 變更所需要的額外 Flink 或 Kafka 資源，方法 1 需要構建整條鏈路，需要更多的 Flink 和 Kafka 資源，因此成本最高。
線上斷流指的是在變更過程中導致下游數據延遲的時長，方法 1 是在數據服務層做切換，幾乎沒有斷流；方法 2 的斷流時長取決于作業從狀態恢復的速度；方法 3 除了狀態恢復，還需要考慮狀態遷移的速度。
等待時長指的是完成整個變更流程需要的時間，前兩個方法都需要重新計算，因此比方法 3 的等待時間更長。
上圖是方法 2 的平臺自動化流程。流程分為七個階段，變更流程執行的時間較長，可能需要幾十分鐘，通過流程條以及圖中每個階段的執行日志可以讓用戶感受到變更的進度和狀態。我們還為用戶做了自動化指標檢查，比如在第 2 個階段的旁路數據回溯中，我們會檢查作業消費 Kafka 的積壓指標，來判斷回溯是否完成，完成后自動制作新邏輯狀態。再比如在第 6 個階段，原作業從旁路作業啟動時會比較 Kafka Offset 指標來比較兩個作業的消費進度，確保線上作業重啟后不會少發數據。
2.3 FlinkSQL 調試繁瑣問題
遇到的第 3 個問題是 FlinkSQL 調試繁瑣，操作步驟多，業務需要創建額外的作業和 Kafka，還要將導出的結果進行存儲。此外，輸入構造復雜，為了針對性地調試某種輸入場景，業務需要寫代碼來構建消息并寫入數據源，甚至需要對多個不同數據源消息到來的順序進行控制。上圖左側可以看到，為了做 FlinkSQL 調試，需要手動搭建一條與線上隔離的調試鏈路，然后寫入 Mock 數據。
解決方案?
針對上述問題的解法是：基于文件調試一鍵化。首先業務在 Web 端可以在線編輯 Mock 數據，Mock 數據是有界的消息序列，它的初始化可以先從線上抽樣，然后再由業務進行修改。業務構建完 Mock 數據后，會將 SQL 作業的 Mock 數據持久化到右側的 S3 文件對象系統上。業務在 Web 端點擊調試，左側發起的調試任務會在與線上隔離的服務器上單進程執行，執行時會從 S3 獲取之前上傳的 Mock 數據，而且可以根據 Mock 數據指定的多源消息之間的到達順序和消息之間的發送間隔來執行，執行完成后會將輸出結果也持久化到 S3，最后在 Web 端查詢 S3 呈現給業務。
更多情況下業務不需要修改 Mock 數據，只需要做抽樣和執行兩步操作。另外我們也支持了一些調試的高級功能，比如支持控制消息的順序和間隔。
上圖是基于以上解法的調試工具。業務會為 SQL 作業創建多個測試用例，其中包括了 Source 的 Mock 數據和 Sink 的預期結果。執行調試后，會檢查所有測試用例的通過情況，通過的條件是要保證結果流 Merge 之后的表與預期表數據一致。
2.4 SQL 作業異常定位問題
第 4 個問題是 FlinkSQL 作業的異常定位。作業異常是指作業消費 Kafka 出現了積壓，為了解決這個問題，需要定位出產生積壓的原因。而定位原因時，歸因的路徑比較復雜，排查門檻比較高。另外由于歸因的路徑缺少系統化的沉淀，定位花費的時間也比較長。隨著 SQL 作業的數量越來越多，如果完全依賴人工排查，工作量將會非常巨大。解決方案
針對上述為的解決方法是實現 SQL 作業的自動化異常診斷。通過 Flink Reporter 上報 SQL 作業的運行指標，并持久化到 TSDB 中用于歷史查詢。同時也會持久化 SQL 作業的運行日志，報警服務會根據規則監控 SQL 作業上報的 Kafka Offset 指標，當消費的 Offset 落后于生產的 Offset 時，會判定位作業發生消費積壓，然后發出報警并下發異常事件，診斷服務會監聽報警服務的異常事件。
異常發生時，根據異常時間窗口內作業日志和作業指標分析異常原因，診斷服務可以通過增加規則來沉淀人工排查的經驗。比如發生了 Restart，就會從日志中根據關鍵字來提取異常信息，未發生 Restart 則會根據反壓指標找出瓶頸節點，然后結合 GC 指標、數據傾斜、火焰圖等來分析瓶頸的原因，最后提出調優建議。
上圖展示了診斷出業務消息臟數據的例子。圖中的運行概況一欄會給出 SQL 作業在每個時間檢查點的診斷情況，綠色表明運行正常，紅色表明作業存在異常，通過這個時間線可以清楚看到異常發生的時間點。診斷結果欄中可以看到異常的原因、詳情和建議。比如在這個事例中，原因是業務消息存在臟數據，在詳情中可以看到導致作業異常的原始消息內容，在建議中會提示業務配置臟數據的處理策略。
03未來規劃
未來，美團實時數倉平臺的規劃主要包括以下兩個方面。
- 首先，是流批一體開發運維，我們即將在實時數倉平臺集成數據湖存儲，并開放 FlinkSQL 的批作業，在存儲和計算層都做到流批統一，提高工作效率。
- 其次，是作業的自動調優，繼續提升作業診斷的準確率以及作業重啟的效率。?
