Apache Flink 在翼支付的實踐應用

摘要：本文整理自翼支付高級開發工程師曹劼、尹春光在 Flink Forward Asia 2021 平臺建設專場的分享。本篇內容主要分爲四個部分：

1. 公司簡介

2. 實踐中的問題

3. 案例實踐

4. 未來規劃

一、公司簡介

翼支付是中國電信的全資子公司，公司主要業務分爲民生繳費、消費購物、金融理財，同時我們依託雲計算、大數據、人工智能等技術手段，賦能線上及線下的商戶。

公司主要的業務板塊分爲數字生活、數字金融及金融科技服務。其中數字生活主要是指常規的支付業務，例如民生繳費，即居民的水電煤氣費繳納等等，同時我們會結合電信聯合推出 5G 的權益套餐；數字金融主要是包含保險、理財、信貸，其中橙分期和企業白條是重點產品；科技服務主要分爲企業徵信及數智業務，企業徵信是指依託現有的雲計算、大數據、人工智能、區塊鏈等核心科技能力，打造專業高效智能的風險管理及徵信科技解決方案。數智業務是指以天翼云爲基礎平臺，重點聚焦 SaaS/PaaS 服務及金融安全服務，打造端到端的金融解決方案。

目前，翼支付的月活用戶數爲 5000 萬 +，存量用戶數 5 個億 +，線上的服務器大約 4000 臺，每日的記錄數爲千億條。

隨着公司的業務規模不斷擴展，我們面臨的業務挑戰也在不斷增多，主要表現在兩個方面：

• 一方面，隨着需求量的不斷增多，採用定製化開發的方式使得應用的數量急劇增加，導致應用難以統一管理，各個業務線的應用向着煙囪式的方向發展，指標口徑和計算不統一，重複的加工會造成能力的浪費；

• 另一方面，某些場景下的單 topic 數據量高達 220 萬 / 秒，同時針對風控等場景，業務響應延遲要求 200 毫秒以內。

針對以上問題，我們從 18 年開始，結合行業的實踐經驗，積極探索建立實時加工體系。在 19 年開始着手構建實時指標加工系統，引入 SparkStreaming 作爲計算引擎。在 20 年初出於對時效性的考慮，我們引入 StructuredStreaming 作爲實時計算引擎。隨着服務的應用不斷增多，我們接收到依賴原子指標的組合的實時決策需求逐漸增多。因此在 20 年 9 月份，我們開始構建實時決策系統，將 FlinkCEP 引入系統中。直到今年 4 月份，爲了解決一些複雜指標的加工需求，我們將 Flink SQL 引入到了指標加工鏈路中。

經過產品的不斷迭代，最終形成了一套企業化的智能決策系統——先鑑平臺。

上圖展示了先鑑平臺的主要功能。首先是實時指標加工。目前我們支持多樣化的數據源，主要包含常用的中間件比如 Kafka 及 Pulsar。同時爲了降低用戶的使用難度，我們提供了 23 種算法模板，也支持 SQL 的定製化加工方式；其次是實時決策。我們支持豐富的規則及規則組的嵌套組合，滿足複雜決策的需求。此外，我們整合了實時、離線及第三方的標籤，爲用戶提供統一的數據查詢服務，同時爲了生產的穩定性，我們提供了全面的監控功能和細粒度資源隔離、熔斷、限流的策略。同時針對實時計算作業的運行狀態，我們對 Source 及 Sink 的數據量和延遲都進行了相關的 Metrics 監控。

上圖展示了先鑑平臺的邏輯架構，主要分爲 4 層。

• 最上層是應用調用方，主要包含智能風控、智能決策、智能營銷系統；

• 往下是實時決策模塊，提供實時決策的功能，其中包含 Web 進行決策的配置及管理，同時提供開發中心進行決策任務的驗證，通過決策核心進行實時的決策；

• 第三層是實時指標加工模塊，通過用戶配置不同的加工方式，錄入到不同的執行引擎，同時整合數據服務，爲用戶提供結果查詢；

• 最下面是數據層，數據源主要包含業務數據、用戶的埋點數據以及集團加工的離線數據。最終根據用戶的配置，將計算結果存儲到相應的 DB。

實時指標加工系統的技術架構圖主要包含三個模塊。前端界面主要負責用戶任務的配置及權限管理，後臺會將用戶配置的信息生成相應的自定義 DSL 語言格式提交給內核，內核根據不同的配置方式，經過 Mode Selector 選擇相應的執行引擎。

如果通過模板的加工方式，則會經過 DSL Parser 進行語法解析，再進行數據的清洗以及算法的計算；如果是 SQL 模式，則只進行 SQL 語法的解析，同時加載用戶的 UDF 及相關配置信息生成相應的任務執行圖交給 Program Deployer 並選擇相應的部署環境進行任務的發佈。

執行環境通過 yarn 進行資源管控，同時 HDFS 負責元數據存儲。

Stream SQL 的功能分爲基礎功能和性能監控功能。

基礎功能主要包括以下幾種：

• SQL 語法校驗。目前支持 Flink SQL 語法，在用戶提交之前先進行 SQL 語法的驗證；

• 沙箱測試。用戶可以預提交任務並進行任務準確性的驗證；

• 支持用戶 udf 函數的加載。

性能監控功能主要包括以下幾種：

• 提供了細粒度的資源配置。社區版本的 Flink SQL 不支持 operator 層級的資源配置，只能使用統一的並行度配置，會導致生產上某個節點壓力過大而造成任務延遲的情況。所以我們通過獲取 Streamgraph 的 JsonPlan 的方式進行各個節點的並行度設置，從而實現細粒度的資源配置；

• 任務狀態監控。我們會監控任務的運行狀態，同時考慮任務延遲以及加工鏈路過長的問題。我們僅僅針對 source 及 sink 的數據流和流量的變化率進行監控，一旦發現變化率異常，會及時反饋給業務用戶，能夠儘早發現業務變化；

• 失敗任務自動恢復。能夠通過獲取最近一次 Checkpoint 進行恢復。同時針對 Checkpoint 週期長的任務，在重啓時考慮恢復時間的問題，我們會在重啓時之前強制進行一次 Savepoint，從而縮短任務恢復時間。

上圖展示了實時指標配置的過程：

• 第一步，配置相應的 Source、Schema 信息或提供數據的 demo 進行自動解析；

• 第二步，選擇數據清洗的方式，這裏提供了幾種簡單的數據清洗邏輯，也支持 SQL 的方式；

• 第三步，選擇計算用的算法模板，也支持算法的嵌套。

上圖展示了 SQL 加工配置的過程。先創建一個任務，包含用戶的資源等參數，然後編寫任務 SQL，最後上線任務並提交給執行環境。

實時決策模塊裏的前端頁面主要負責決策任務的配置及用戶權限管理，並將任務提交給後端。後端會通過 Zookeeper 將上線的策略發佈到相應的決策節點。每一個執行節點都有一個 ZK Watcher，用於監聽策略的狀態，通過 RuleLoader 加載策略並通過 RuleCompiler 對策略進行編譯，最後交給 Flink CEP 進行決策執行。最終將決策的結果存儲到 DB 或中間件。

決策配置的過程首先需要創建一個任務，然後配置相應的規則以及規則的組合，最後通過開發中心進行任務的試運行，驗證決策的準確性。

二、實踐中的問題

在實踐過程中，我們也遇到了很多挑戰，總結起來有如下幾個方面：

業務 State 數據一致性、指標重複計算問題、動態規則配置以及全鏈路監控監控問題。

首先是指標作業升級過程中，通過指標引擎配置的 job State 數據一致性問題。早期指標作業是通過手動開發，部分業務 State 存儲在 HDFS 中，指標引擎配置的 job 沒有單獨管理業務 State 的數據，老的任務遷移到平臺過程中就會遇到數據一致性問題。

解決思路是擴展老的計算程序，讀取全量 State 數據存儲到外部，然後停止老任務。指標引擎配置的作業從指定的 offset 進行數據計算，然後從外部存儲補齊原有的指標數據。

上圖展示了作業升級的流程。Task 在 open function 的時候讀取業務 State 數據存儲到外部。如果是 Keyed State，則 State 接口無法獲取當前 task 的所有 State 數據，需要將 State 對象進行向下類型強轉，然後獲取所有 State 數據指標引擎。作業通過配置指定對應的 offset，通過從外部補齊數據的方式進行指標計算，從而完成數據恢復。

其次是指標作業在不斷新增過程中存在的痛點，多個作業重複消費同一個 Kafka 導致上游消費壓力大以及指標重複計算的問題。

針對以上痛點，我們的解決方法是對所有作業進行統一優化，對所有消息源進行統一預清洗，按照業務過程分發到對應的數據域 Topic 中。對指標進行統一的口徑管理，保證指標不重複計算。目前沒有對實時指標進行分層處理，主要爲了避免在計算鏈路過長從而影響業務的時效性。

第三，Flink CEP 存在的問題。實時決策的模塊是通過 Flink CEP 進行規則匹配，最初是通過程序編碼的方式實現規則的匹配，然而隨着規則越來越多，不便於維護，開發成本也隨之增加。Flink CEP 無法進行動態的規則配置以及多個規則並行決策。

針對上述問題，我們對 Flink CEP 進行了擴展開發來解決規則動態配置以及多個規則決策的問題。

上圖展示了 Flink CEP 擴展開發的邏輯架構。用戶通過 RuleManager 配置規則並將規則變更事件發佈到 Zookeeper 中，RuleListener 監聽到事件的變更後，若是新增規則，則會通過 groovy 動態語言編譯生成 RulePattern 實例。隨着規則的增多，CEP operator 線程處理效率會下降，需要通過把規則分組綁定到對應的 Worker 上來加速規則處理。CEP operator 線程接收到事件後會分發給所有 Worker，Worker 線程處理完後通過隊列發佈到 CEP operator 線程，最後發佈到下游。

最後是數據全鏈路監控的問題。數據流從收集端經過 Flume 傳輸，再到消息中心指標計算，然後發佈到下游的實時決策，不允許大量的數據丟失以及數據延遲。

基於以上訴求，需要對整體數據鏈路進行監控，採用 prometheus + grafana 進行 metrics 的收集以及告警。這裏主要針對 Flume 消息中間件進行消息堆積以及丟失的監控。Flink 指標計算主要監控運行狀態以及背壓情況，下游監控 CEP 決策的時間。對數據鏈路的監控能夠幫助運維快速定位並解決線上的問題。

三、案例實踐

上圖展示了先鑑平臺的工作方式。

首先，上游的用戶行爲和業務事件通過數據通道傳輸到先鑑平臺，業務方負責配置實時指標和業務規則，當業務事件通過指標計算結果觸發了業務規則，先鑑平臺隨即將結果推送到下游的消息中心，通過各業務系統觸達到用戶。比如用戶訪問理財首頁時，如果 30 分鐘內未進行產品申購，就會根據用戶的資質給他發送對應的推送短信。

四、未來規劃

未來，我們計劃在以下幾個方面進行持續探索：

• 第一，數據庫增量採集的方案統一。目前 MySQL 的採集是使用 Canal 實現的，未來我們規劃使用 Flink CDC 來針對 Oracle 和 MySQL 進行統一的增量採集；

• 第二，離線實時的批流融合。目前離線數倉通過 Spark SQL 計算，實時數倉使用 Flink SQL 計算，維護兩套元數據以及不同的指標口徑使得日常工作負荷很大，因此我們希望使用 Flink 來完成統一的批流計算；

• 第三，Flink 作業自動擴容縮容。目前 Flink 無法進行自動擴容縮容，早晚流量變化較大，會導致較多的資源浪費，計算能力不足的時候只能通過人工進行作業擴容。我們希望基於 Flink 來實現自動擴容，降低運維成本。

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/Rv_cR6cDcLwQWT50qYqjEw