一種實(shí)時(shí)數(shù)據(jù)庫(kù)
數(shù)據(jù)同步方案及實(shí)現(xiàn)
神州信息
董志
1.
概述
變化數(shù)據(jù)捕獲簡(jiǎn)稱CDC（Change Data Capture），可以識(shí)別提取從上次提取之后發(fā)生變化的數(shù)據(jù)，在廣義的概念上，只要能捕獲數(shù)據(jù)變更的技術(shù)，我們都可以稱為CDC。通常我們說(shuō)的 CDC技術(shù)主要面向數(shù)據(jù)庫(kù)的變更，是一種用于捕獲數(shù)據(jù)庫(kù)中數(shù)據(jù)變更的技術(shù)。CDC的兩種模式：
（1）同步：同步CDC主要是采用觸發(fā)器記錄新增數(shù)據(jù)，基本能夠做到實(shí)時(shí)增量提取。
（2）異步：異步CDC通過(guò)分析已經(jīng)提交的日志記錄來(lái)得到增量數(shù)據(jù)信息，有一定的延時(shí)，是本文采用的模式。
1.1. 應(yīng)用場(chǎng)景
（1）數(shù)據(jù)同步，用于備份、容災(zāi)。
（2）數(shù)據(jù)分發(fā)，一個(gè)數(shù)據(jù)源分發(fā)給多個(gè)下游。
（3）數(shù)據(jù)采集，面向數(shù)據(jù)倉(cāng)庫(kù)/數(shù)據(jù)湖的ETL數(shù)據(jù)集成。
1.2. 主流的實(shí)現(xiàn)機(jī)制
（1）基于查詢的CDC
a)離線調(diào)度查詢作業(yè)，批處理。
b)無(wú)法保障數(shù)據(jù)一致性。
c)不保障實(shí)時(shí)性。
（2）基于日志的CDC
a)實(shí)時(shí)消費(fèi)日志，流處理。
b)保障數(shù)據(jù)一致性。
c)提供實(shí)時(shí)數(shù)據(jù)。
2.
方案對(duì)比
主流開(kāi)源CDC方案對(duì)比如下圖所示，主要通過(guò)監(jiān)控各數(shù)據(jù)庫(kù)的事務(wù)日志達(dá)到監(jiān)控?cái)?shù)據(jù)變化的目的，根據(jù)對(duì)比采用Flink CDC 方案。
圖：多種CDC技術(shù)對(duì)比
(1) DataX 不支持增量同步，Canal 不支持全量同步。雖然兩者都是非常流行的數(shù)據(jù)同步工具，但在場(chǎng)景支持上仍不完善。
(2) 在全量+增量一體化同步方面，只有Flink CDC、Debezium、Oracle Goldengate 支持較好。
(3) 在架構(gòu)方面，Apache Flink 是一個(gè)非常優(yōu)秀的分布式流處理框架，因此Flink CDC 作為Apache Flink 的一個(gè)組件具有非常靈活的水平擴(kuò)展能力。而DataX 和Canal 是個(gè)單機(jī)架構(gòu)，在大數(shù)據(jù)場(chǎng)景下容易面臨性能瓶頸的問(wèn)題。
(4) 在數(shù)據(jù)加工的能力上，CDC 工具是否能夠方便地對(duì)數(shù)據(jù)做一些清洗、過(guò)濾、聚合，甚至關(guān)聯(lián)操作？Flink CDC 依托強(qiáng)大的Flink SQL 流式計(jì)算能力，可以非常方便地對(duì)數(shù)據(jù)進(jìn)行加工。而 Debezium 等則需要通過(guò)復(fù)雜的 Java 代碼才能完成，使用門(mén)檻比較高。
(5) 另外，在生態(tài)方面，這里指的是上下游存儲(chǔ)的支持。Flink CDC 上下游非常豐富，支持對(duì)接MySQL、PostgreSQL 等數(shù)據(jù)源，還支持寫(xiě)入到TiDB、HBase、Kafka、Hudi 等各種存儲(chǔ)系統(tǒng)中，也支持靈活的自定義connector。
因此，不論從性能還是適用范圍上，F(xiàn)link CDC 都可以作為最佳選擇。Flink CDC Connectors是Apache Flink的一組source連接器，使用變更數(shù)據(jù)捕獲 (CDC) 從不同的數(shù)據(jù)庫(kù)中獲取變更，F(xiàn)link CDC連接器集成了Debezium作為引擎來(lái)捕獲數(shù)據(jù)變化，所以它可以充分發(fā)揮Debezium的能力。目前連接器支持的數(shù)據(jù)庫(kù)有：MySQL（5.6+）、PostgreSQL（9.6+）、MongoDB（3.6+）、Oracle（11+）、TiDB（5.1.x+）、SQL Server（2012+）和Oceanbase（3.1.x+）。
3.
數(shù)據(jù)庫(kù)事務(wù)日志
目前支持的關(guān)系型數(shù)據(jù)庫(kù)包括：MySQL、Oracle、PostgreSQL、SQL Server，主要采用基于WAL日志方式進(jìn)行數(shù)據(jù)變化監(jiān)聽(tīng)。下面介紹各關(guān)系型數(shù)據(jù)庫(kù)的日志類型：
1. MySQL
（1）Error log錯(cuò)誤日志記錄了MySQL Server運(yùn)行過(guò)程中所有較為嚴(yán)重的警告和錯(cuò)誤信息，以及MySQL Server每次啟動(dòng)和關(guān)閉的詳細(xì)信息。
（2）Binary log二進(jìn)制日志，記錄著數(shù)據(jù)庫(kù)發(fā)生的各種事務(wù)信息。
（3）Update log更新日志是MySQL在較老版本上使用的，其功能跟Bin log類似，只不過(guò)不是以二進(jìn)制格式記錄，而是以簡(jiǎn)單文本格式記錄內(nèi)容。
（4）Query log查詢?nèi)罩居涗汳ySQL中所有的query。
（5）Slow query log慢查詢?nèi)罩居涗浀木褪菆?zhí)行時(shí)間較長(zhǎng)的query。
（6）InnoDB redo log，InnoDB是一個(gè)事務(wù)安全的存儲(chǔ)引擎，其事務(wù)安全性主要就是通過(guò)在線redo日志和記錄在表空間中的undo信息來(lái)保證的。
2. Oracle
（1）系統(tǒng)報(bào)警日志alert.log。
（2）跟蹤日志(用戶和進(jìn)程) trace.log。
（3）重做日志。
a. 在線重做日志：又稱聯(lián)機(jī)重做日志，指Oracle以SQL腳本的形式實(shí)時(shí)記錄數(shù)據(jù)庫(kù)的數(shù)據(jù)更新，換句話說(shuō)，實(shí)時(shí)保存已執(zhí)行的SQL腳本到在線日志文件中（按特定的格式）。
b. 歸檔重做日志：指當(dāng)條件滿足時(shí)，Oracle將在線重做日志以文件形式保存到硬盤(pán)（持久化）。
3. PostgreSQL
（1）pg_log文件夾中的日志一般用來(lái)記錄服務(wù)器與DB的狀態(tài)，如各種Error信息，定位慢查詢SQL，數(shù)據(jù)庫(kù)的啟動(dòng)關(guān)閉信息，發(fā)生checkpoint過(guò)于頻繁等的告警信息等。
（2）pg_xlog文件夾中的日志是記錄的PostgreSQL的WAL信息，也就是一些事務(wù)日志信息(transaction log)，記錄著數(shù)據(jù)庫(kù)發(fā)生的各種事務(wù)信息。
（3）pg_clog文件夾存儲(chǔ)的也是事務(wù)日志文件，但與pg_xlog不同的是它記錄的是事務(wù)的元數(shù)據(jù)(metadata)，這個(gè)日志告訴我們哪些事務(wù)完成了，哪些沒(méi)有完成。
4. SQL Server
（1）交易日志（Transaction logs），是針對(duì)數(shù)據(jù)庫(kù)改變所做的記錄，它可以記錄針對(duì)數(shù)據(jù)庫(kù)的任何操作，并將記錄結(jié)果保存在獨(dú)立的文件中。對(duì)于任何每一個(gè)交易過(guò)程，交易日志都有非常全面的記錄，根據(jù)這些記錄可以將數(shù)據(jù)文件恢復(fù)成交易前的狀態(tài)。
4.
功能實(shí)現(xiàn)
1. 整體架構(gòu)
整體架構(gòu)如下圖所示，首先各源端數(shù)據(jù)庫(kù)需要開(kāi)啟相應(yīng)的事務(wù)日志，F(xiàn)link CDC 任務(wù)會(huì)監(jiān)聽(tīng)各數(shù)據(jù)庫(kù)的事務(wù)變化日志，然后對(duì)日志數(shù)據(jù)進(jìn)行處理，最后將數(shù)據(jù)進(jìn)行傳輸：
（1）通過(guò)訂閱發(fā)布方式將消息發(fā)送到Redis 的Channel 中，通知消費(fèi)者數(shù)據(jù)庫(kù)中的數(shù)據(jù)發(fā)生了變化。
（2）以流的方式存儲(chǔ)到Kafka 的 Topic中，供下游程序進(jìn)行消費(fèi)。
（3）抽取到其他關(guān)系型數(shù)據(jù)庫(kù)中，實(shí)現(xiàn) ETL 功能。
圖：整體架構(gòu)圖
2. 數(shù)據(jù)格式
由于 Flink CDC 內(nèi)部集成了 Debezium 組件，通過(guò) Debezium 進(jìn)行數(shù)據(jù)采集，所以數(shù)據(jù)格式同 Debezium，監(jiān)聽(tīng)到的數(shù)據(jù)格式如下圖所示，after 代表變化后的數(shù)據(jù)；source 代表源端的數(shù)據(jù)庫(kù)相關(guān)信息，包括 Debezium 版本號(hào)、連接器類型、數(shù)據(jù)庫(kù)名、表名等；op 代表操作的類型，此處為讀操作。
圖：數(shù)據(jù)格式
3. 事務(wù)日志開(kāi)啟
（1） MySQL 開(kāi)啟Bin Log 日志
在 my.cnf 里面加上如下配置，重啟服務(wù)。
查看是否開(kāi)啟Bin Log日志 show variables like 'log_%';
（2） Oracle 開(kāi)啟歸檔日志
啟用歸檔日志：
檢查歸檔日志是否啟用：
啟動(dòng)補(bǔ)充日志記錄：
4. 具體代碼實(shí)現(xiàn)
DataStream方式監(jiān)聽(tīng) MySQL 數(shù)據(jù)庫(kù)實(shí)現(xiàn)：
DataStream方式監(jiān)聽(tīng) Oracle 數(shù)據(jù)庫(kù)實(shí)現(xiàn)：
Flink SQL方式監(jiān)聽(tīng) MySQL 數(shù)據(jù)庫(kù)實(shí)現(xiàn)：
自定義反序列化器：
自定義Redis Sink：
