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