汽车之家基于 Flink 的数据传输平台的设计与实践
數據接入與傳輸作為打通數據系統與業務系統的一道橋梁，是數據系統與架構中不可或缺的一個重要部分。數據傳輸系統穩定性和準確性，直接影響整個數據系統服務的 SLA 和質量。此外如何提升系統的易用性，保證監控服務并降低系統維護成本，優雅應對災難等問題也十分重要。
本文介紹了汽車之家實時計算團隊利用 Flink 和 Flink 實時平臺構建數據傳輸 SDK 和傳輸平臺并不斷完善的實踐經驗與總結。內容包括：
一、背景與需求
汽車之家（下稱之家）作為一家數據智能驅動的公司，天然存在著對數據的各種復雜需求，之家的數據系統負責支撐這些業務需求的開展。數據傳輸系統，作為其中一環，承擔了各類數據導入分發的需求，支持用戶訂閱數據變更。隨著支撐的業務擴增與需求的增加。原來的接入系統暴露出了一定的問題和不足：
- 缺乏有效的任務與信息管理機制，依賴人工進行任務的管理和運維，信息的統計
- 接入程序資源使用浪費，缺乏彈性
- 針對 DDL 變更問題，不能很好的處理，必要時需要人工介入
- 傳輸系統依賴的組件比較多，比如 Zookeeper，Redis 等
- 代碼的技術債累積，代碼維護成本變高
針對上述問題，我們決定開發一套新的數據傳輸和分發系統，一舉解決上述問題。
二、技術選型與設計 —— Why Flink?
在開展新系統的開發工作之前，我們分析的可選的方案思路大體分三種:
我們規約出以下主要設計使用目標：
- 架構設計上要運維管理是友好的，提供高可用以及故障恢復策略，支持異地多活
- 架構設計上要提供強數據準確性，至少承諾 at-least-once 語義
- 架構設計上要對擴縮容是友好的，可以按需分配資源
- 功能設計上要全面的監控覆蓋和完善的報警機制，支持元數據信息管理
- 功能設計上要對實時計算是友好的(1)
- 功能設計上要能完全防御 DDL 變更帶來的問題
此外，在性能指標上，接入系統的延時和吞吐至少要滿足所有業務常規狀態下的需求。
(1) 指與實時計算平臺整合的能力方案設計與對比
依照設計思路和目標，我們整理了方案主要功能的對比表格：
(1)Flink 自帶高可用和故障恢復，實時計算平臺在此基礎上提供更強的高可用服務(2)良好的編碼 + flink 機制即可實現 Exactly-Once
(3)實時計算平臺自帶任務部署管理能力
(4)實時計算平臺自帶完備的監控和管理
經過討論，大家一致決定基于Flink進行新的傳輸平臺的開發：
我們的 MVP 版本開發完成大約只花費了不到 3 周的時間，POC 的結果完全符合預期的性能要求和功能要求。
三、數據傳輸系統的設計架構
從邏輯層面來看，之家的實時數據傳輸平臺分為 3 部分：
- 數據傳輸程序
- 接入任務信息管理模塊
- 任務執行 Runtime 模塊
在實現上：
- 數據傳輸程序是由固定的 Flink Jar 和 Flink SQL Codegen Service 生成的SQL Task 組成
- 管理模塊作為一個微服務，負責與 Flink 平臺組件通信，完成必要的任務管理和信息管理
- 執行層直接依賴 Flink 平臺和 Flink 平臺的集群
組件架構與交互邏輯
傳輸系統涉及到的組件和交互如圖所示：
AutoDTS 即為傳輸系統的任務信息管理模塊，AutoStream Core 為 Flink 實時平臺核心系統，Jar Service 是 Flink 相關 SDK Jar 儲存管理服務，Metastore 為 Flink 平臺的元數據管理系統，Flink Client 是我們自己封裝的 Submit Client，支持以 Restful 方式向 YARN/K8S 上提交作業。
AutoDTS 前端直接與用戶進行交互，完成用戶對任務信息的修改和任務生命周期的操作。AutoDTS 將任務信息處理后與 Flink 平臺交互，每一個數據傳輸任務對應Flink平臺唯一一個任務，同時，部分任務信息被 AutoDTS 處理，會直接在 Metastore 上完成對應流表的創建。用戶直接申請并使用該 Flink 流表，進行 SQL 任務的開發。
針對不同的傳輸任務，AutoDTS 會委托 Core System 組織任務參數和 SQL 邏輯，并從 Jar Service 加載不同的 SDK Jar 提交到 Client 去執行，對于基于 SQL Codegen 的傳輸任務，Flink SQL Codegen Service 會將任務參數組織整合翻譯成可執行的 Flink SQL 任務，通過 SQL 任務，我們可以直接復用平臺 SQL SDKs，執行 SQL 作業。
正如前文提到的，我們最大限度復用已有組件和服務，大大降低了開發的周期。
傳輸任務類型與構成
之家的數據傳輸任務分為兩種類型，接入任務與分發任務。
- 接入任務，負責從數據源實時接入 Changelog Stream 并處理成統一的格式寫入 Kafka 中，每個表只會對用唯一個接入程序，作為公共數據資產，被下游程序進行使用和消費
- 分發任務， 負責讀取公共的 Kafka 數據，并將數據寫入指定的存儲中，用戶根據自己的需求去使用，擁有分發任務的所有權
如圖所示，接入的數據源主要有 3 種，除了 Mysql 和 SqlServer，我們還支持了 TiDB 的 Changelog(TiCDC)接入 Java Client 相關邏輯,并將我們的代碼貢獻到了 TiDB 社區 [1]；對于分發端，通過解析用戶的任務配置，從而進行 SQL codegen 生成 Flink SQL 代碼執行。
四、基于 Flink 的 Binlog 接入 SDK
在這些接入和分發 SDK 中，Binlog 接入 SDK 是比較有難度的一個，下面我們以 Binlog 接入 SDK 為例，剖析接入 SDK 的主體設計思路和開發過程。
Stage 拆解
依照 Flink 經典的 Source->Transformation->Sink，Binlog 接入任務也拆分為這三個Stage：
Binlog Source
Binlog Source 的樸素開發思路：創建一個 BinaryLogClient 并持續 fetchBinlogEvent 并進行簡單的轉換處理后發送到下游。在既定的設計目標中，以下問題需要認真思考：
對于問題1，考慮到 Binlog Stream 的特殊性，我們要求 Source 的并行度為且僅能為1。且在絕大部分情況下，從 BinaryLogClient fetch BinlogEvent 不會是性能瓶頸。我們只要保證 BinaryLogClient 與 BinlogSourceFunction 的生命周期一致，二者通過有界的阻塞隊列鏈接，分別充當生產者和消費者，同時 BinlogSourceFunction 對 BinlogEvent 盡可能少的進行邏處理，讓 BinlogSourceFunction 的負擔盡量減輕，從而提升 Source 階段的性能即可。
而對于問題 2、3，則需要從 Binlog 的特性和格式來分析。眾所周知，BinlogEvent 攜帶了唯一的 BinlogPosition。BinlogPosition 是全序的，我們可以在 trigger Checkpoint 的時候，對當前的 BinlogPosition 進行記錄。但是僅僅是記錄這個是不夠的，如果記錄了數據位置，那么下次從 Checkpoint 恢復的時候，是從當條記錄開始還是當條記錄的下一條記錄開始呢？另一方面，我們希望發送的按照一個完整的 transaction 去發送數據給下游而非從事務中間截斷發送。這里，我們就要用到 BinlogEvent 的一種特定事件——TransactionEnd 事件。
我們這里先來解決問題 2，我們要求 BinlogSourceFunction 只使用 TransactionEnd 事件的 BinlogPosition 來更新位點保存到狀態中，由于 TransactionEnd 事件不是 DML 事件，不會導致下游生成數據，所以就不需要考慮之前提到的問題。
而問題3的解決需要和 Flink的Checkpoint 機制進行聯動。我們當時使用的 Flink 版本是 1.9.x。在 Source 端，需要通過 CheckpointLock 來讓 Source 和 Checkpoint trigger 進行配合。雖然在理解和使用上有一定的壁壘，但是 CheckppointLock 機制恰恰幫助我們達成了問題 3 的目標。我們保證了 Source 只有拿到 lock 才發送數據給下游，只有在完成一次 transaction 的數據發送后才 unlock，這樣就保證了 2 個 checkpoint 之間必定是完整的 𝒳（ 𝒳 ∈ N ）次 transaction 的數據。另一方面，我們減小了 checkpoint trigger 的間隔（200ms~500ms），減少了 checkpoint 間的數據 transaction 的數量，加快數據 commit 的速度。
UnifiedFormatTransform
就如名字描述的，UnifiedFormatTransform 的作用是將數據轉換為統一制定的數據格式。
相較于 Binlog Source 階段，UnifiedFormatTransform 階段不用太過擔心性能問題，良好的編碼和水平垂直擴容能力可以應付絕大部分性能需求。但是有一個重要的問題亟待解決，就是前面提到的功能設計目標：完全防御 DDL 帶來的問題。
DDL 問題在數據同步/傳輸中一直是一個比較棘手的問題，帶來的麻煩包括不限于，數據解析失敗/錯誤，程序失敗/重啟，且恢復的成本往往很高。而其實解決這個問題的核心思路也很簡單，就是在程序中就地解析 DDL 并處理 Schema 變化。為了實現這個功能，我們需要完成以下幾個步驟：
- 內嵌 Parser，用于解析 DDL SQL
- 解析出現的所有 DDL，根據解析的 DDL 內容更新內置的 Schema，并更新到 Flink 狀態中
- 生成 DDL 對應的數據發送到下游
我們這實現上參考了 Maxwell [2] 的做法，內嵌了 Antlr4 的 Mysql 文法的 g4 文件，然后自定義 listener 來完成對 Schema 的更新和 DDL 數據的生成，然后 Schema 會在 Checkpoint 觸發時被保存到狀態中。
完成了就地解決 DDL 的功能后，不論是簡單的 Alter Table，還是復雜的 Online DDL，接入程序都可以順利解決，利用狀態從斷點恢復，也不會出現 Schema 異常的問題。
Kafka Sink
Kafka Sink 階段主要是將轉換好的數據寫入 Kafka中。Flink 原生為 Kafka Sink 賦予了 Exactly-Once 的能力，而我們也將這個功能利用起來，和 Source 一起，提供了開箱即用的端到端 Exactly-Once 解決方案。我們保證了 Source 按照完整的 Mysql Transaction 發送數據，同時 Sink 按照完整的 Mysql Transaction 將數據寫入Kafka，對于 Transaction 敏感的場景，我們可以開啟 Transactional 消費模式，來完成強 transaction 語義（而非最終一致性）的數據處理。
其他優化
此外我們還做了一些優化功能：
- gtid 支持與一鍵主從切換
- 程序運行信息定期備份到外部存儲
- Binlog 同步任務相關的監控指標覆蓋
五、平臺使用
用戶在傳輸平臺，只需要完成必要配置的設定，即可完成傳輸任務的創建和數據的使用，比較簡單。
接入任務
對于接入任務，正如我們前文提到的, 接入任務產生的數據會被作為公共資產。所以用戶只需要查詢需求的表的數據是否已經接入，如果已經接入，則可以直接申請使用，否則發起一次表接入申請，審批通過后會由系統自動進行操作。
分發任務
對于分發作業，需要用戶進行創建，以 Iceberg 分發任務為例：
■ 字段篩選
選擇出分發作業使用的已經接入到平臺的數據源表字段
在選擇一些任務的運行配置（如資源，運行環境）后，就可以創建并運行一個分發任務，我們可以看到對應唯一一個 Flink 平臺任務 ID：
此外，我們還提供了豐富的監控查詢，元數據信息查詢等功能，充分利用了實時計算平臺的已有組件，實現了傳輸系統與實時計算系統的緊密結合。
六、總結與展望
實踐證明，我們選擇基于 Flink 進行輸出傳輸系統的開發，是個明智且正確的決定。在最小的開發成本下，從功能和效率及可維護性上，完全解決了之前遺留的問題，全面提升了之家接入/分發/數據訂閱的效率和用戶體驗，也提升了我們在數據傳輸方面的技術能力。
最近我們在數據湖方向投入了較多的精力，傳輸系統目前也已經初步支持數據接入數據湖，未來希望可以不斷完善相關功能，大幅提升數據湖數據接入的能力，支持用戶一鍵入湖，加強整個數據體系的整合。
另一方面，我們看到 Flink 新版本提供了許多新功能新工具。例如 FLIP-27 Source 和 OperatorCoordinator，我們希望可以借由這兩個全新的機制和工具，繼續優化我們的代碼，拓展相關功能。對于新推出 Upsert-Kafka，我們已經開始嘗試在 Flink 計算平臺上進行初步的開發和整合，希望之后將 Upsert-Kafka 與傳輸系統打通，繼續擴展與豐富實時計算和傳輸的業務場景！
參考資料：
[1] https://github.com/pingcap/ticdc/pull/804
[2] https://github.com/zendesk/maxwell
作者介紹：
劉首維，本科畢業于大連理工大學，Apache Flink Contributor，Scala/Akka 重度愛好者，19年加入汽車之家負責實時計算平臺和數據傳輸平臺數據的開發和維護。
原文鏈接：https://developer.aliyun.com/article/783578?
版權聲明：本文內容由阿里云實名注冊用戶自發貢獻，版權歸原作者所有，阿里云開發者社區不擁有其著作權，亦不承擔相應法律責任。具體規則請查看《阿里云開發者社區用戶服務協議》和《阿里云開發者社區知識產權保護指引》。如果您發現本社區中有涉嫌抄襲的內容，填寫侵權投訴表單進行舉報，一經查實，本社區將立刻刪除涉嫌侵權內容。 與50位技術專家面對面20年技術見證，附贈技術全景圖總結
以上是生活随笔為你收集整理的汽车之家基于 Flink 的数据传输平台的设计与实践的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 饿了么EMonitor演进史
- 下一篇： 融合趋势下基于 Flink Kylin