摘要：MRSCDL 是 FusionInsight MRS 推出的一種資料實時同步服務，旨在将傳統 OLTP 資料庫中的事件資訊捕捉并實時推送到大資料産品中去，本文檔會詳細為大家介紹 CDL 的整體架構以及關鍵技術。
本文分享自華為雲社群《MRS CDL架構設計與實作》，作者：rujia01。
1、前言
MRS CDL 是 FusionInsight MRS 推出的一種資料實時同步服務，旨在将傳統 OLTP 資料庫中的事件資訊捕捉并實時推送到大資料産品中去，本文檔會詳細為大家介紹 CDL 的整體架構以及關鍵技術。
2、CDL 的概念
MRS CDL（Change Data Loader）是一款基于 Kafka Connect 的 CDC 資料同步服務，可以從多種 OLTP 資料源捕獲資料，如 Oracle、MySQL、PostgreSQL 等，然後傳輸給目标存儲，該目标存儲可以大資料存儲如 HDFS，OBS，也可以是實時資料湖 Hudi 等。
2.1 什麼是 CDC？
CDC(Change DataCapture)是一種通過監測資料變更（新增、修改、删除等）而對變更的資料進行進一步處理的一種設計模式，通常應用在資料倉庫以及和資料庫密切相關的一些應用上，比如資料同步、備份、審計、ETL 等。
CDC 技術的誕生已經有些年頭了，二十多年前，CDC 技術就已經用來捕獲應用資料的變更。CDC 技術能夠及時有效的将消息同步到對應的數倉中，并且幾乎對目前的生産應用不産生影響。如今，大資料應用越來越普遍，CDC 這項古老的技術重新煥發了生機，對接大資料場景已經是 CDC 技術的新使命。
目前業界已經有許多成熟的 CDC to 大資料的産品，如：OracleGoldenGate(for Kafka)、 Ali/Canal、Linkedin/Databus、Debezium/Debezium 等等。
2.2 CDL 支援的場景
MRS CDL 吸收了以上成熟産品的成功經驗，采用 Oracle LogMinner 和開源的 Debezium 來進行 CDC 事件的捕捉，借助 Kafka 和 Kafka Connect 的高并發，高吞吐量，高可靠架構進行任務的部署。
現有的 CDC 産品在對接大資料場景時，基本都會選擇将資料同步到消息隊列 Kafka 中。MRS CDL 在此基礎上進一步提供了資料直接入湖的能力，可以直接對接 MRS HDFS 和 Huawei OBS 以及 MRS Hudi、ClickHouse 等，解決資料的最後一公裡問題。
表 1 MRS CDL 支援的場景
3、CDL 的架構
作為一個 CDC 系統，能夠從源目标抽取資料并且傳輸到目标存儲中去是基本能力，在此基礎上，靈活、高性能、高可靠、可擴充、可重入、安全是 MRS CDL 着重考慮的方向，是以，CDL 的核心設計原則如下：
- 系統結構必須滿足可擴充性原則，支援在不損害現有系統功能的前提下添加新的源和目标資料存儲。
- 架構設計應當滿足不同角色間的業務側重點分離
- 在合理的情況下減少複雜性和依賴性，最大限度的降低架構、安全性、韌性方面的風險。
- 需要滿足插件式的客戶需求，提供通用的插件能力，使得系統靈活、易用、可配置。
- 業務安全，避免橫向越權和資訊洩露。
3.1 架構圖/角色介紹
圖 1 CDL 架構
MRS CDL 包含 CDL Service 和 CDL Connector 兩個角色，他們各自的職能如下：
- CDL Service：負責任務的管理和排程，提供統一的 API 接口，同時監測整個 CDL 服務的健康狀态。
- CDL Connector：本質上是 Kafka Connect 的 Worker 程序，負責真實 Task 的運作，在 Kafka Connect 高可靠、高可用、可擴充的特性基礎上增加了心跳機制來協助 CDL Service 完成叢集的健康監測。
3.2 為什麼選擇 Kafka？
我們将 Apache Kafka 與 Flume 和 Nifi 等各種其他選項進行了比較，如下表所示：
表 1 架構比較
對于 CDC 系統，Kafka 有足夠的優勢來支撐我們做出選擇。同時，Kafka Connect 的架構完美契合 CDC 系統：
- 并行 - 對于一個資料複制任務，可以通過拆解成多個子任務并且并行運作來提高吞吐率。
- 保序 - Kafka 的 partition 機制可以保證在一個 partition 内資料嚴格有序，這樣有助于我們實作資料完整性。
- 可擴充 - Kafka Connect 在叢集中分布式的運作 Connector。
- 易用 - 對 Kafka 的接口進行了抽象，提升了易用性。
- 均衡 - Kafka Connect 自動檢測故障，并在剩餘程序上根據各自負載重新進行均衡排程。
- 生命周期管理 – 提供完善的 Connector 的生命周期管理能力。
4、MRS CDL 關鍵技術
圖 2 CDL 關鍵技術
4.1 CDL Job
MRS CDL 對業務進行了上層的抽象，通過引入 CDL Job 的概念來定義一個完整的業務流程。在一個 Job 中，使用者可以選擇資料源和目标存儲類型，并且可以篩選要複制的資料表。
在 Job 結構的基礎上，MRS CDL 提供執行 CDL Job 的機制，在運作時，使用 Kafka Connect Source Connector 結合日志複制技術将 CDC 事件從源資料存儲捕獲到 Kafka，然後使用 Kafka Connect Sink Connector 從 Kafka 提取資料，在應用各種轉換規則後将最終結果推送到目标存儲。
提供定義表級和列級映射轉換的機制，在定義 CDL Job 的過程中可以指定轉換規則。
4.2 DataComparison
MRS CDL 提供一種特殊的 Job，用于進行資料一緻性對比。使用者可以選擇源和目标資料存儲架構，從源和目标架構中選擇各種比較對進行資料比較，以確定資料在源和目标資料存儲中一緻。
圖 3 Data Comparison 抽象視圖
MRS CDL 提供了專用的 Rest API 來運作 Data Compare Job，并且提供如下能力：
- 提供多樣的資料比較算法，如行雜湊演算法，非主鍵列比較等。
- 提供專門的查詢接口，可以查詢同步報表，展示目前 Compare 任務的執行明細。
- 提供實時的基于源和目标存儲的修複腳本，一鍵修複不同步資料。
如下是 Data Compare Job 執行流程：
圖 4 Data Compare Job 執行和檢視流程
4.3 SourceConnectors
MRS CDL 通過 Kafka Connect SDK 建立各種源連接配接器，這些連接配接器從各種資料源捕獲 CDC 事件并推送到 Kafka。CDL 提供專門的 Rest API 來管理這些資料源連接配接器的生命周期。
4.3.1 OracleSource Connector
Oracle SourceConnector 使用 Oracle RDBMS 提供的 Log Miner 接口從 Oracle 資料庫捕獲 DDL 和 DML 事件。
圖 5 Log Miner 抓取資料示意圖
在處理 DML 事件時，如果表中存在 BOLB/CLOB 列，CDL 同樣可以提供支援。對于 BOLB 列的處理，關鍵點處理如下：
- 當 insert/update 操作發生時，會觸發一系列的 LOB_WRITE 操作。
- LOB_WRITE 用于将檔案加載到 BLOB 字段中。
- 每個 LOB_WRITE 隻能寫入 1KB 資料。
- 對于一個 1GB 的圖檔檔案，我們會整理全部的 100 萬個 LOB_WRITE 操作中的二進制資料，然後合并成一個對象。我們會把這個對象存儲到 Huawei OBS 中，最終在寫入 Kafka 的 message 中給出該對象在 OBS 中的位置。
對于 DDL 事件的捕獲，我們建立單獨的會話來持續跟蹤。目前支援的 DDL 語句如下：
表 2 支援的 DDL 語句
4.3.2 MYSQLSource Connector
MYSQL 的 Binary Log（Bin Log）檔案順序記錄了所有送出到資料庫的操作，包括了對表結構的變更和對表資料的變更。MYSQL Source Connector 通過讀取 Bin Log 檔案，生産 CDC 事件并送出到 Kafka 的 Topic 中。
MYSQL SourceConnector 主要支援的功能場景有：
- 捕獲 DML 事件，并且支援并行處理所捕獲的 DML 事件，提升整體性能
- 支援表過濾
- 支援配置表和 Topic 的映射關系
- 為了保證 CDC 事件的絕對順序，我們一般要求一張表隻對應一個 Partition，但是，MYSQL Source Connector 仍然提供了寫入多 Partition 的能力，來滿足某些需要犧牲消息保序性來提升性能的場景
- 提供基于指定 Bin Log 檔案、指定位置或 GTID 來重新開機任務的能力，保證異常場景下資料不丢失
- 支援多種複雜資料類型
- 支援捕獲 DDL 事件
4.3.3 PostgreSQLSource Connector
PostgreSQL 的邏輯解碼特性允許我們解析送出到事務日志的變更事件，這需要通過輸出插件來處理這些變更。PostgreSQLSource Connector 使用 pgoutput 插件來完成這項工作。pgoutput 插件是 PostgreSQL 10+提供的标準邏輯解碼插件，無需安裝額外的依賴包。
PostgreSQL SourceConnector 和 MYSQL Source Connector 除了部分資料類型的差別外其他功能基本一緻。
4.4 SinkConnectors
MRS 提供多種 Sink Connector，可以從 Kafka 中拉取資料并推送到不同的目标存儲中。現在支援的 Sink Connector 有：
- HDFS Sink Connector
- OBS Sink Connector
- Hudi Sink Connector
- ClickHouse Sink Connector
- Hive Sink Connector
其中 Hudi Sink Connector 和 ClickHouse SinkConnector 也支援通過 Flink/Spark 應用來排程運作。
4.5 表過濾
當我們想在一個 CDL Job 中同時捕獲多張表的變更時，我們可以使用通配符（正規表達式）來代替表名，即允許同時捕獲名稱滿足規則的表的 CDC 事件。當通配符（正規表達式）不能嚴格比對目标時，就會出現多餘的表被捕獲。為此，CDL 提供表過濾功能，來輔助通配符模糊比對的場景。目前 CDL 同時支援白名單和黑名單兩種過濾方式。
4.6 統一資料格式
MRS CDL 對于不同的資料源類型如 Oracle、MYSQL、PostgreSQL 采用了統一的消息格式存儲在 Kafka 中，後端消費者隻需解析一種資料格式來進行後續的資料處理和傳輸，避免了資料格式多樣導緻後端開發成本增加的問題。
4.7 任務級的日志浏覽
通常境況下，一個 CDL Connector 會運作多個 Task 線程來進行 CDC 事件的抓取，當其中一個 Task 失敗時，很難從海量的日志中抽取出強相關的日志資訊，來進行進一步的分析。
為了解決如上問題，CDL 規範了 CDL Connector 的日志列印，并且提供了專用的 REST API，使用者可以通過該 API 一鍵擷取指定 Connector 或者 Task 的日志檔案。甚至可以指定起止時間來進一步縮小日志查詢的範圍。
4.8 監控
MRS CDL 提供 REST API 來查詢 CDL 服務所有核心部件的 Metric 資訊，包括服務級、角色級、執行個體級以及任務級。
4.9 應用程式錯誤處理
在業務運作過程中，常常會出現某些消息無法發送到目标資料源的情況，我們把這種消息叫做錯誤記錄。在 CDL 中，出現錯誤記錄的場景有很多種，比如：
- Topic 中的消息體與特定的序列化方式不比對，導緻無法正常讀取
- 目标存儲中并不存在消息中所存儲的表名稱，導緻消息無法發送到目标端
為了處理這種問題，CDL 定義了一種“dead letter queue”，專門用于存儲運作過程中出現的錯誤記錄。本質上“dead letter queue”是由 Sink Connector 建立的特定的 Topic，當出現錯誤記錄時，由 Sink Connector 将其發往“dead letterqueue”進行存儲。
同時，CDL 提供了 REST API 來供使用者随時查詢這些錯誤記錄進行進一步分析，并且提供 Rest API 可以允許使用者對這些錯誤記錄進行編輯和重發。
圖 6 CDL Application Error Handling
5、性能
CDL 使用了多種性能優化方案來提高吞吐量：
- Task 并發
我們利用 Kafka Connect 提供的任務并行化功能，其中 Connect 可以将作業拆分為多個任務來并行複制資料，如下所示：
圖 7 Task 并發
- 使用 Executor 線程并行化執行任務
由于 Log Miner，Bin Log 等資料複制技術的限制，我們的 Source Connector 隻能順序的捕獲 CDC 事件，是以，為了提高性能，我們将這些 CDC 事件先緩存到記憶體隊列中，然後使用 Executor 線程并行的處理它們。這些線程會先從内部隊列中讀取資料，然後處理并且推送到 Kafka 中。
圖 8 Executor 線程并發
6、總結
MRS CDL 是資料實時入湖場景下重要的一塊拼圖，我們仍然需要在資料一緻性、易用性、多元件對接以及性能提升等場景需要進一步擴充和完善，在未來能夠更好的為客戶創造價值。