柚子快報激活碼778899分享：大數(shù)據(jù) FlinkCDC詳解
柚子快報激活碼778899分享：大數(shù)據(jù) FlinkCDC詳解
1、FlinkCDC是什么
1.1 CDC是什么
CDC是Chanage Data Capture（數(shù)據(jù)變更捕獲）的簡稱。其核心原理就是監(jiān)測并捕獲數(shù)據(jù)庫的變動（例如增刪改），將這些變更按照發(fā)生順序捕獲，將捕獲到的數(shù)據(jù)，寫入數(shù)據(jù)庫種如神策數(shù)據(jù)的核心kudu、doris、mysql、kakfa等。
1.2 CDC的實現(xiàn)方式
1.2.1 基于查詢的CDC
離線調(diào)度查詢作業(yè)，批處理。把一張表同步到其他系統(tǒng)，每次通過查詢?nèi)カ@取表中最新的數(shù)據(jù)無法保障數(shù)據(jù)一致性，查的過程中有可能數(shù)據(jù)已經(jīng)發(fā)生了多次變更；不保障實時性，基于離線調(diào)度存在天然的延遲。
1.2.2 基于日志的CDC
實時消費日志，流處理，例如 MySQL 的 binlog 日志完整記錄了數(shù)據(jù)庫中的變更，可以把 binlog 文件當(dāng)作流的數(shù)據(jù)源；保障數(shù)據(jù)一致性，因為 binlog 文件包含了所有歷史變更明細(xì)；保障實時性，因為類似 binlog 的日志文件是可以流式消費的，提供的是實時數(shù)據(jù)。
1.2.3 常見的開源的CDC方案比較
1.2.4 個人對于CDC領(lǐng)域的一些淺見
其實對于CDC領(lǐng)域在數(shù)倉行業(yè)中很常見，無論是離線數(shù)倉也好還是實時數(shù)倉也好，或者說是業(yè)務(wù)系統(tǒng)也好，例如京東就是使用CDC方案來同步優(yōu)惠卷的。其實在很多的CDC的同步方案中，大部分公司其實選用的是第一種，查詢同步方案，為什么這么做呢，很多人可能會問，實時同步不好嗎，我想說的是實時的CDC太復(fù)雜，雖然一致性不高，但是其實運營或者其他人員并不需要這么高的實時性，可能某些領(lǐng)域需要，當(dāng)然也有很多的表結(jié)構(gòu)設(shè)計沒有update_time字段，這樣的話如果同步一張表，可能會有點麻煩，但是并非是不能同步，如果數(shù)據(jù)量不大的話，或者有其他自增鍵的話會很方便，但是如果沒有的話就會很麻煩，也可以做，可以做整行的md5這里我就不一一贅述了，在進(jìn)行查詢cdc同步的一些情況。日志cdc呢，其實根本原理就是監(jiān)控類似于mysql的binlog。可以讓整個數(shù)據(jù)的增刪改，進(jìn)行捕獲，從而可以達(dá)到兩個數(shù)據(jù)的一致性，當(dāng)然這個一致性并不是實時的，哪怕是mysql的主從都有可能延遲，更別提咱們監(jiān)控binlog了，當(dāng)然這種延遲幾乎很少見，業(yè)務(wù)也不會發(fā)現(xiàn)，這種CDC雖然聽上去很好，但是實現(xiàn)較為困難，限制比較大，例如下游的數(shù)據(jù)源要支持改，不像離線可以用拉鏈表來解決。但是這種方式真的很好，如果開發(fā)人員和架構(gòu)設(shè)計人員以及數(shù)據(jù)設(shè)計人員的設(shè)計比較好，這種方式效果是最棒的，我司的mysql同步器就支持這兩種方式，根據(jù)使用人員的喜好來進(jìn)行選擇。
2、Flink CDC的原理
2.1 1.x Flink CDC
Flink1.x的cdc依賴于Debezium組件，debezium為了保證數(shù)據(jù)的一致性，在全量讀取時，會加鎖。 此時呢會分為全局鎖權(quán)限和無全局鎖權(quán)限。
那么為什么debezium為什么要這么做呢，要加上全局鎖呢，因為數(shù)據(jù)一致性問題，這就涉及到數(shù)據(jù)庫的全局鎖和表鎖了，數(shù)據(jù)庫的全局鎖，以mysql為例，全局鎖就是對整個數(shù)據(jù)庫實例加鎖。MySQL 提供了一個加全局讀鎖的方法，命令是Flush tables with read lock (FTWRL)。 當(dāng)你需要讓整個庫處于只讀狀態(tài)的時候，可以使用這個命令，之后其他線程的以下語句會被阻塞：數(shù)據(jù)更新語句（數(shù)據(jù)的增刪改）、數(shù)據(jù)定義語句（包括建表、修改表結(jié)構(gòu)等）和更新類事務(wù)的提交語句。一般全局鎖的使用場景在數(shù)據(jù)庫備份上，當(dāng)然如果主庫加鎖的話，會導(dǎo)致一些問題。例如加鎖后，這個數(shù)據(jù)庫實例無法更新，業(yè)務(wù)基本就停止了。從庫呢，也不能從binlog拉取數(shù)據(jù)，這就導(dǎo)致了主從延遲，假如有的業(yè)務(wù)使用的是從庫的話就會出現(xiàn)問題。當(dāng)然全局鎖有問題，那么不加鎖會導(dǎo)致什么問題呢，數(shù)據(jù)不一致問題： 比如手機(jī)卡，購買套餐信息
這里分為兩張表 u_acount (用于余額表)，u_pricing (資費套餐表)
步驟:
1. u_account 表中數(shù)據(jù) 用戶A 余額：300
u_pricing 表中數(shù)據(jù) 用戶A 套餐：空
2. 發(fā)起備份，備份過程中先備份u_account表，備份完了這個表，這個時候u_account 用戶余額是300
3. 這個時候套用戶購買了一個資費套餐100，餐購買完成，寫入到u_print套餐表購買成功，備份期間的數(shù)據(jù)。
4. 備份完成
可以看到備份的結(jié)果是，u_account 表中的數(shù)據(jù)沒有變， u_pricing 表中的數(shù)據(jù) 已近購買了資費套餐100.
哪這時候用這個備份文件來恢復(fù)數(shù)據(jù)的話，用戶A 賺了100 ，用戶是不是很舒服啊。但是你得想想公司利益啊?！ ?也就是說，不加鎖的話，備份系統(tǒng)備份的得到的庫不是一個邏輯時間點，這個數(shù)據(jù)是邏輯不一致的。
當(dāng)然mysql的備份工具，mysqldump可以在備份的時候支持更新，基于MVCC的機(jī)制。MVCC （Multiversion Concurrency Control），多版本并發(fā)控制。顧名思義，MVCC 是通過數(shù)據(jù)行的 多個版本 管理來實現(xiàn)數(shù)據(jù)庫的 并發(fā)控制。這項技術(shù)使得在InnoDB的事務(wù)隔離級別下執(zhí)行 一致性讀操 作有了保證。換言之，就是為了查詢一些正在被另一個事務(wù)更新的行，并且可以看到它們被更新之前的值，這樣在做查詢的時候就不用等待另一個事務(wù)釋放鎖。 不再深入解釋mysql的核心機(jī)制了。 表鎖是什么呢，顧名思義就是鎖住了整張表。在加表鎖的表上，無法進(jìn)行DDL、DML操作。當(dāng)然在mysql5.5以后，有一個表鎖是MDL，MDL不需要顯示的使用，在訪問一個表的時候會被自動加上。MDL 的作用是，保證讀寫的正確性。你可以想象一下，如果一個查詢正在遍歷一個表中的數(shù)據(jù)，而執(zhí)行期間另一個線程對這個表結(jié)構(gòu)做變更，刪了一列，那么查詢線程拿到的結(jié)果跟表結(jié)構(gòu)對不上，肯定是不行的。因此，在 MySQL 5.5 版本中引入了 MDL，當(dāng)對一個表做增刪改查操作的時候，加 MDL讀鎖；當(dāng)要對表做結(jié)構(gòu)變更操作的時候，加 MDL 寫鎖。
讀鎖之間不互斥，因此你可以有多個線程同時對一張表增刪改查。讀寫鎖之間、寫鎖之間是互斥的，用來保證變更表結(jié)構(gòu)操作的安全性。因此，如果有兩個線程要同時給一個表加字段，其中一個要等另一個執(zhí)行完才能開始執(zhí)行。
MDL鎖有一些問題，假如在多個讀session中進(jìn)行更改表結(jié)構(gòu)操作的話，可能會卡死。
這個就是debezium在flink1.x中的應(yīng)用。
2.2 2.x Flink CDC
Flink 2.x不僅引入了增量快照讀取機(jī)制，還帶來了一些其他功能的改進(jìn)。以下是對Flink 2.x的主要功能的介紹：
增量快照讀?。篎link 2.x引入了增量快照讀取機(jī)制，這是一種全新的數(shù)據(jù)讀取方式。該機(jī)制支持并發(fā)讀取和以chunk為粒度進(jìn)行checkpoint。在增量快照讀取過程中，F(xiàn)link首先根據(jù)表的主鍵將其劃分為多個塊（chunk），然后將這些塊分配給多個讀取器并行讀取數(shù)據(jù)。這一機(jī)制極大地提高了數(shù)據(jù)讀取的效率。 精確一次性處理：Flink 2.x引入了Exactly-Once語義，確保數(shù)據(jù)處理結(jié)果的精確一次性。MySQL CDC 連接器是Flink的Source連接器，可以利用Flink的checkpoint機(jī)制來確保精確一次性處理。 動態(tài)加表：Flink 2.x支持動態(tài)加表，通過使用savepoint來復(fù)用之前作業(yè)的狀態(tài)，解決了動態(tài)加表的問題。 無主鍵表的處理：Flink 2.x對無主鍵表的讀取和處理進(jìn)行了優(yōu)化。在無主鍵表中，F(xiàn)link可以通過一些額外的字段來識別數(shù)據(jù)記錄的唯一性，從而實現(xiàn)準(zhǔn)確的數(shù)據(jù)讀取和處理。
對于Flink 2.x的CDC方案呢，可以理解為全量讀取時，在劃分chunk塊的時候，采用了查詢讀，他是將主鍵進(jìn)行切分的。默認(rèn)一個chunk8096條數(shù)據(jù)，知道這些就可以了。 2.x的 Flink cdc實現(xiàn)較為復(fù)雜，這里就不一一贅述了。
3、FlinkCDC的使用
3.1 導(dǎo)入依賴
3.2 代碼實操
import com.ververica.cdc.connectors.mysql.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.DebeziumSourceFunction;
import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class FlinkCDC {
public static void main(String[] args) throws Exception {
//1.獲取Flink執(zhí)行環(huán)境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
//通過FlinkCDC構(gòu)建SourceFunction
DebeziumSourceFunction
.hostname("hadoop102")
.port(3306)
.username("root")
.password("123456")
.databaseList("cdc_test") //監(jiān)控的數(shù)據(jù)庫
.tableList("cdc_test.user_info") //監(jiān)控的數(shù)據(jù)庫下的表
.deserializer(new StringDebeziumDeserializationSchema())//反序列化
.startupOptions(StartupOptions.initial())
.build();
DataStreamSource
//3.數(shù)據(jù)打印
dataStreamSource.print();
//4.啟動任務(wù)
env.execute("FlinkCDC");
}
}
4、Flink CDC輸出數(shù)據(jù)解析
4.1 數(shù)據(jù)的數(shù)據(jù)結(jié)構(gòu)
flink cdc的輸出結(jié)果大概可以分為 before、after、 before代表變更前數(shù)據(jù)，after代表變更后數(shù)據(jù)。
還有個op，這個op代表的是事務(wù)的操作： r：讀取歷史 d：刪除 c：創(chuàng)建 u：更新
柚子快報激活碼778899分享：大數(shù)據(jù) FlinkCDC詳解
推薦閱讀
本文內(nèi)容根據(jù)網(wǎng)絡(luò)資料整理，出于傳遞更多信息之目的，不代表金鑰匙跨境贊同其觀點和立場。
轉(zhuǎn)載請注明，如有侵權(quán)，聯(lián)系刪除。
ePRICE價格獵人
