MRS CDL架構(gòu)設(shè)計與實現(xiàn)

1 前言

MRS CDL是FusionInsight MRS推出的一種數(shù)據(jù)實時同步服務(wù)，旨在將傳統(tǒng)OLTP數(shù)據(jù)庫中的事件信息捕捉并實時推送到大數(shù)據(jù)產(chǎn)品中去，本文檔會詳細(xì)為大家介紹CDL的整體架構(gòu)以及關(guān)鍵技術(shù)。

2 CDL的概念

MRS CDL（Change Data Loader）是一款基于Kafka Connect的CDC數(shù)據(jù)同步服務(wù)，可以從多種OLTP數(shù)據(jù)源捕獲數(shù)據(jù)，如Oracle、MySQL、PostgreSQL等，然后傳輸給目標(biāo)存儲，該目標(biāo)存儲可以大數(shù)據(jù)存儲如HDFS，OBS，也可以是實時數(shù)據(jù)湖Hudi等。

2.1 什么是CDC？

CDC(Change Data Capture)是一種通過監(jiān)測數(shù)據(jù)變更（新增、修改、刪除等）而對變更的數(shù)據(jù)進(jìn)行進(jìn)一步處理的一種設(shè)計模式，通常應(yīng)用在數(shù)據(jù)倉庫以及和數(shù)據(jù)庫密切相關(guān)的一些應(yīng)用上，比如數(shù)據(jù)同步、備份、審計、ETL等。

CDC技術(shù)的誕生已經(jīng)有些年頭了，二十多年前，CDC技術(shù)就已經(jīng)用來捕獲應(yīng)用數(shù)據(jù)的變更。CDC技術(shù)能夠及時有效的將消息同步到對應(yīng)的數(shù)倉中，并且?guī)缀鯇Ξ?dāng)前的生產(chǎn)應(yīng)用不產(chǎn)生影響。如今，大數(shù)據(jù)應(yīng)用越來越普遍，CDC這項古老的技術(shù)重新煥發(fā)了生機(jī)，對接大數(shù)據(jù)場景已經(jīng)是CDC技術(shù)的新使命。

當(dāng)前業(yè)界已經(jīng)有許多成熟的CDC to大數(shù)據(jù)的產(chǎn)品，如：Oracle GoldenGate(for Kafka)、 Ali/Canal、Linkedin/Databus、Debezium/Debezium等等。

2.2 CDL支持的場景

MRS CDL吸收了以上成熟產(chǎn)品的成功經(jīng)驗，采用Oracle LogMinner和開源的Debezium來進(jìn)行CDC事件的捕捉，借助Kafka和Kafka Connect的高并發(fā)，高吞吐量，高可靠框架進(jìn)行任務(wù)的部署。

現(xiàn)有的CDC產(chǎn)品在對接大數(shù)據(jù)場景時，基本都會選擇將數(shù)據(jù)同步到消息隊列Kafka中。MRS CDL在此基礎(chǔ)上進(jìn)一步提供了數(shù)據(jù)直接入湖的能力，可以直接對接MRS HDFS和Huawei OBS以及MRS Hudi、ClickHouse等，解決數(shù)據(jù)的最后一公里問題。

場景

數(shù)據(jù)源

目標(biāo)存儲

實時數(shù)據(jù)湖分析

Oracle

Huawei OBS, MRS HDFS, MRS Hudi, MRS ClickHouse, MRS Hive

MySQL

Huawei OBS, MRS HDFS, MRS Hudi, MRS ClickHouse, MRS Hive

PostgreSQL

Huawei OBS, MRS HDFS, MRS Hudi, MRS ClickHouse, MRS Hive

表1 MRS CDL支持的場景

3 CDL的架構(gòu)

作為一個CDC系統(tǒng)，能夠從源目標(biāo)抽取數(shù)據(jù)并且傳輸?shù)侥繕?biāo)存儲中去是基本能力，在此基礎(chǔ)上，靈活、高性能、高可靠、可擴(kuò)展、可重入、安全是MRS CDL著重考慮的方向，因此，CDL的核心設(shè)計原則如下：

系統(tǒng)結(jié)構(gòu)必須滿足可擴(kuò)展性原則，支持在不損害現(xiàn)有系統(tǒng)功能的前提下添加新的源和目標(biāo)數(shù)據(jù)存儲。

架構(gòu)設(shè)計應(yīng)當(dāng)滿足不同角色間的業(yè)務(wù)側(cè)重點分離

在合理的情況下減少復(fù)雜性和依賴性，最大限度的降低架構(gòu)、安全性、韌性方面的風(fēng)險。

需要滿足插件式的客戶需求，提供通用的插件能力，使得系統(tǒng)靈活、易用、可配置。

業(yè)務(wù)安全，避免橫向越權(quán)和信息泄露。

3.1 架構(gòu)圖/角色介紹

圖1 CDL架構(gòu)

MRS CDL包含CDL Service和CDL Connector兩個角色，他們各自的職能如下：

CDL Service：負(fù)責(zé)任務(wù)的管理和調(diào)度，提供統(tǒng)一的API接口，同時監(jiān)測整個CDL服務(wù)的健康狀態(tài)。

CDL Connector：本質(zhì)上是Kafka Connect的Worker進(jìn)程，負(fù)責(zé)真實Task的運行，在Kafka Connect高可靠、高可用、可擴(kuò)展的特性基礎(chǔ)上增加了心跳機(jī)制來協(xié)助CDL Service完成集群的健康監(jiān)測。

3.2 為什么選擇Kafka？

我們將Apache Kafka與Flume和Nifi等各種其他選項進(jìn)行了比較，如下表所示：

Flume

Nifi

Kafka

優(yōu)點

基于配置的Agent架構(gòu)

-

Source、Channel、Sink模型

有許多開箱即用的處理器

背壓機(jī)制

處理任意大小的消息

支持MiNifi Agent來收集數(shù)據(jù)

支持邊緣層數(shù)據(jù)流

可擴(kuò)展、分布式、高容錯、高吞吐量的消息傳遞系統(tǒng)

背壓機(jī)制

無數(shù)據(jù)丟失

Kafka Connect支持Source、Sink模型

超過50種可用的Connector

消息保序

低耦合

缺點

存在數(shù)據(jù)丟失的場景

沒有數(shù)據(jù)備份

數(shù)據(jù)大小限制

沒有背壓機(jī)制

沒有數(shù)據(jù)復(fù)制

脆弱的容錯機(jī)制

不支持消息保序

可擴(kuò)展性較差

消息大小限制

表1 框架比較

對于CDC系統(tǒng)，Kafka有足夠的優(yōu)勢來支撐我們做出選擇。同時，Kafka Connect的架構(gòu)完美契合CDC系統(tǒng)：

并行 - 對于一個數(shù)據(jù)復(fù)制任務(wù)，可以通過拆解成多個子任務(wù)并且并行運行來提高吞吐率。

保序 - Kafka的partition機(jī)制可以保證在一個partition內(nèi)數(shù)據(jù)嚴(yán)格有序，這樣有助于我們實現(xiàn)數(shù)據(jù)完整性。

可擴(kuò)展 - Kafka Connect在集群中分布式的運行Connector。

易用 - 對Kafka的接口進(jìn)行了抽象，提升了易用性。

均衡 - Kafka Connect自動檢測故障，并在剩余進(jìn)程上根據(jù)各自負(fù)載重新進(jìn)行均衡調(diào)度。

生命周期管理 – 提供完善的Connector的生命周期管理能力。

4 MRS CDL關(guān)鍵技術(shù)

圖2 CDL關(guān)鍵技術(shù)

4.1 CDL Job

MRS CDL對業(yè)務(wù)進(jìn)行了上層的抽象，通過引入CDL Job的概念來定義一個完整的業(yè)務(wù)流程。在一個Job中，用戶可以選擇數(shù)據(jù)源和目標(biāo)存儲類型，并且可以篩選要復(fù)制的數(shù)據(jù)表。

在Job結(jié)構(gòu)的基礎(chǔ)上，MRS CDL提供執(zhí)行CDL Job的機(jī)制，在運行時，使用Kafka Connect Source Connector結(jié)合日志復(fù)制技術(shù)將CDC事件從源數(shù)據(jù)存儲捕獲到Kafka，然后使用Kafka Connect Sink Connector從Kafka提取數(shù)據(jù)，在應(yīng)用各種轉(zhuǎn)換規(guī)則后將最終結(jié)果推送到目標(biāo)存儲。

提供定義表級和列級映射轉(zhuǎn)換的機(jī)制，在定義CDL Job的過程中可以指定轉(zhuǎn)換規(guī)則。

4.2 Data Comparison

MRS CDL提供一種特殊的Job，用于進(jìn)行數(shù)據(jù)一致性對比。用戶可以選擇源和目標(biāo)數(shù)據(jù)存儲架構(gòu)，從源和目標(biāo)架構(gòu)中選擇各種比較對進(jìn)行數(shù)據(jù)比較，以確保數(shù)據(jù)在源和目標(biāo)數(shù)據(jù)存儲中一致。

圖3 Data Comparison抽象視圖

MRS CDL提供了專用的Rest API來運行Data Compare Job，并且提供如下能力：

提供多樣的數(shù)據(jù)比較算法，如行哈希算法，非主鍵列比較等。

提供專門的查詢接口，可以查詢同步報表，展示當(dāng)前Compare任務(wù)的執(zhí)行明細(xì)。

提供實時的基于源和目標(biāo)存儲的修復(fù)腳本，一鍵修復(fù)不同步數(shù)據(jù)。

如下是Data Compare Job執(zhí)行流程：

圖4 Data Compare Job執(zhí)行和查看流程

4.3 Source Connectors

MRS CDL通過Kafka Connect SDK創(chuàng)建各種源連接器，這些連接器從各種數(shù)據(jù)源捕獲CDC事件并推送到Kafka。CDL提供專門的Rest API來管理這些數(shù)據(jù)源連接器的生命周期。

4.3.1 Oracle Source Connector

Oracle Source Connector使用Oracle RDBMS提供的Log Miner接口從Oracle數(shù)據(jù)庫捕獲DDL和DML事件。

圖5 Log Miner抓取數(shù)據(jù)示意圖

在處理DML事件時，如果表中存在BOLB/CLOB列，CDL同樣可以提供支持。對于BOLB列的處理，關(guān)鍵點處理如下：

當(dāng)insert/update操作發(fā)生時，會觸發(fā)一系列的LOB_WRITE操作。

LOB_WRITE用于將文件加載到BLOB字段中。

每個LOB_WRITE只能寫入1KB數(shù)據(jù)。

對于一個1GB的圖片文件，我們會整理全部的100萬個LOB_WRITE操作中的二進(jìn)制數(shù)據(jù)，然后合并成一個對象。我們會把這個對象存儲到Huawei OBS中，最終在寫入Kafka的message中給出該對象在OBS中的位置。

對于DDL事件的捕獲，我們創(chuàng)建單獨的會話來持續(xù)跟蹤。當(dāng)前支持的DDL語句如下：

No

DDL語句

示例

1

CREATE TABLE

CREATE TABLE TEST ( EMPID INT PRIMARY KEY, ENAME VARCHAR2(10))

2

ALTER TABLE ... ADD ( )

ALTER TABLE TEST ADD ( SALARY NUMBER)

3

ALTER TABLE ... DROP COLUMN ...

ALTER TABLE TEST DROP (SALARY)

4

ALTER TABLE ... MODIFY ( ...

ALTER TABLE TEST MODIFY SALARY INT

5

ALTER ... RENAME...

ALTER TABLE TEST RENAME TO CUSTOMER

6

DROP ...

DROP TABLE TEST

7

CREATE UNIQUE INDEX ...

CREATE UNIQUE INDEX TESTINDEX ON TEST (EMPID, ENAME)

8

DELETE INDEX …

Delete existing index

表2 支持的DDL語句

4.3.2 MYSQL Source Connector

MYSQL的Binary Log（Bin Log）文件順序記錄了所有提交到數(shù)據(jù)庫的操作，包括了對表結(jié)構(gòu)的變更和對表數(shù)據(jù)的變更。MYSQL Source Connector通過讀取Bin Log文件，生產(chǎn)CDC事件并提交到Kafka的Topic中。

MYSQL Source Connector主要支持的功能場景有：

捕獲DML事件，并且支持并行處理所捕獲的DML事件，提升整體性能

支持表過濾

支持配置表和Topic的映射關(guān)系

為了保證CDC事件的絕對順序，我們一般要求一張表只對應(yīng)一個Partition，但是，MYSQL Source Connector仍然提供了寫入多Partition的能力，來滿足某些需要犧牲消息保序性來提升性能的場景

提供基于指定Bin Log文件、指定位置或GTID來重啟任務(wù)的能力，保證異常場景下數(shù)據(jù)不丟失

支持多種復(fù)雜數(shù)據(jù)類型

支持捕獲DDL事件

4.3.3 PostgreSQL Source Connector

PostgreSQL的邏輯解碼特性允許我們解析提交到事務(wù)日志的變更事件，這需要通過輸出插件來處理這些變更。PostgreSQL Source Connector使用pgoutput插件來完成這項工作。pgoutput插件是PostgreSQL 10+提供的標(biāo)準(zhǔn)邏輯解碼插件，無需安裝額外的依賴包。

PostgreSQL Source Connector和MYSQL Source Connector除了部分?jǐn)?shù)據(jù)類型的區(qū)別外其他功能基本一致。

4.4 Sink Connectors

MRS提供多種Sink Connector，可以從Kafka中拉取數(shù)據(jù)并推送到不同的目標(biāo)存儲中。現(xiàn)在支持的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應(yīng)用來調(diào)度運行。

4.5 表過濾

當(dāng)我們想在一個CDL Job中同時捕獲多張表的變更時，我們可以使用通配符（正則表達(dá)式）來代替表名，即允許同時捕獲名稱滿足規(guī)則的表的CDC事件。當(dāng)通配符（正則表達(dá)式）不能嚴(yán)格匹配目標(biāo)時，就會出現(xiàn)多余的表被捕獲。為此，CDL提供表過濾功能，來輔助通配符模糊匹配的場景。當(dāng)前CDL同時支持白名單和黑名單兩種過濾方式。

4.6 統(tǒng)一數(shù)據(jù)格式

MRS CDL對于不同的數(shù)據(jù)源類型如Oracle、MYSQL、PostgreSQL采用了統(tǒng)一的消息格式存儲在Kafka中，后端消費者只需解析一種數(shù)據(jù)格式來進(jìn)行后續(xù)的數(shù)據(jù)處理和傳輸，避免了數(shù)據(jù)格式多樣導(dǎo)致后端開發(fā)成本增加的問題。

4.7 任務(wù)級的日志瀏覽

通常境況下，一個CDL Connector會運行多個Task線程來進(jìn)行CDC事件的抓取，當(dāng)其中一個Task失敗時，很難從海量的日志中抽取出強(qiáng)相關(guān)的日志信息，來進(jìn)行進(jìn)一步的分析。

為了解決如上問題，CDL規(guī)范了CDL Connector的日志打印，并且提供了專用的REST API，用戶可以通過該API一鍵獲取指定Connector或者Task的日志文件。甚至可以指定起止時間來進(jìn)一步縮小日志查詢的范圍。

4.8 監(jiān)控

MRS CDL提供REST API來查詢CDL服務(wù)所有核心部件的Metric信息，包括服務(wù)級、角色級、實例級以及任務(wù)級。

4.9 應(yīng)用程序錯誤處理

在業(yè)務(wù)運行過程中，常常會出現(xiàn)某些消息無法發(fā)送到目標(biāo)數(shù)據(jù)源的情況，我們把這種消息叫做錯誤記錄。在CDL中，出現(xiàn)錯誤記錄的場景有很多種，比如：

Topic中的消息體與特定的序列化方式不匹配，導(dǎo)致無法正常讀取

目標(biāo)存儲中并不存在消息中所存儲的表名稱，導(dǎo)致消息無法發(fā)送到目標(biāo)端

為了處理這種問題，CDL定義了一種“dead letter queue”，專門用于存儲運行過程中出現(xiàn)的錯誤記錄。本質(zhì)上“dead letter queue”是由Sink Connector創(chuàng)建的特定的Topic，當(dāng)出現(xiàn)錯誤記錄時，由Sink Connector將其發(fā)往“dead letter queue”進(jìn)行存儲。

同時，CDL提供了REST API來供用戶隨時查詢這些錯誤記錄進(jìn)行進(jìn)一步分析，并且提供Rest API可以允許用戶對這些錯誤記錄進(jìn)行編輯和重發(fā)。

圖6 CDL Application Error Handling

5 性能

CDL使用了多種性能優(yōu)化方案來提高吞吐量：

Task并發(fā)

我們利用Kafka Connect提供的任務(wù)并行化功能，其中Connect可以將作業(yè)拆分為多個任務(wù)來并行復(fù)制數(shù)據(jù)，如下所示：

圖7 Task并發(fā)

使用Executor線程并行化執(zhí)行任務(wù)

由于Log Miner，Bin Log等數(shù)據(jù)復(fù)制技術(shù)的限制，我們的Source Connector只能順序的捕獲CDC事件，因此，為了提高性能，我們將這些CDC事件先緩存到內(nèi)存隊列中，然后使用Executor線程并行的處理它們。這些線程會先從內(nèi)部隊列中讀取數(shù)據(jù)，然后處理并且推送到Kafka中。

圖8 Executor線程并發(fā)

6 總結(jié)

MRS CDL是數(shù)據(jù)實時入湖場景下重要的一塊拼圖，我們?nèi)匀恍枰跀?shù)據(jù)一致性、易用性、多組件對接以及性能提升等場景需要進(jìn)一步擴(kuò)展和完善，在未來能夠更好的為客戶創(chuàng)造價值。

EI企業(yè)智能 FusionInsight Hadoop 大數(shù)據(jù)

版權(quán)聲明：本文內(nèi)容由網(wǎng)絡(luò)用戶投稿，版權(quán)歸原作者所有，本站不擁有其著作權(quán)，亦不承擔(dān)相應(yīng)法律責(zé)任。如果您發(fā)現(xiàn)本站中有涉嫌抄襲或描述失實的內(nèi)容，請聯(lián)系我們jiasou666@gmail.com 處理，核實后本網(wǎng)站將在24小時內(nèi)刪除侵權(quán)內(nèi)容。