您好,登錄后才能下訂單哦!
如何理解Apache Flink CDC原理與使用,相信很多沒有經(jīng)驗的人對此束手無策,為此本文總結(jié)了問題出現(xiàn)的原因和解決方法,通過這篇文章希望你能解決這個問題。
Flink在1.11版本中新增了CDC的特性,簡稱 改變數(shù)據(jù)捕獲。名稱來看有點亂,我們先從之前的數(shù)據(jù)架構(gòu)來看CDC的內(nèi)容。
以上是之前的mysql binlog日志處理流程,例如canal監(jiān)聽binlog把日志寫入到kafka中。而Apache Flink實時消費Kakfa的數(shù)據(jù)實現(xiàn)mysql數(shù)據(jù)的同步或其他內(nèi)容等。拆分來說整體上可以分為以下幾個階段。
整體的處理鏈路較長,需要用到的組件也比較多。Apache Flink CDC可以直接從數(shù)據(jù)庫獲取到binlog供下游進行業(yè)務計算分析。簡單來說鏈路會變成這樣也就是說數(shù)據(jù)不再通過canal與kafka進行同步,而flink直接進行處理mysql的數(shù)據(jù)。節(jié)省了canal與kafka的過程。
Flink 1.11中實現(xiàn)了mysql-cdc與postgre-CDC,也就是說在Flink 1.11中我們可以直接通過Flink來直接消費mysql,postgresql的數(shù)據(jù)進行業(yè)務的處理。
首先需要保證mysql數(shù)據(jù)庫開啟了binlog。未開啟請查閱相關(guān)資料進行binlog的啟用。自建默認是不開啟binlog的。
DROP TABLE IF EXISTS `t_test`;
CREATE TABLE `t_test` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`ip` varchar(255) DEFAULT NULL,
`size` bigint(20) DEFAULT NULL
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=183 DEFAULT CHARSET=utf8mb4;
<dependency>
<groupId>com.alibaba.ververica</groupId>
<artifactId>flink-connector-mysql-cdc</artifactId>
<version>1.1.0</version>
<scope>compile</scope>
</dependency>
def main(args: Array[String]): Unit = {
val envSetting = EnvironmentSettings.newInstance()
.useBlinkPlanner()
.inStreamingMode()
.build()
val env = StreamExecutionEnvironment.getExecutionEnvironment
val tableEnv = StreamTableEnvironment.create(env, envSetting)
val sourceDDL =
"CREATE TABLE test_binlog (" +
" id INT NOT NULl," +
" ip STRING," +
" size INT" +
") WITH (" +
"'connector' = 'mysql-cdc'," +
"'hostname' = 'localhost'," +
"'port' = '3306'," +
"'username' = 'root'," +
"'password' = 'cain'," +
"'database-name' = 'test'," +
"'table-name' = 't_test'" +
")"
// 輸出目標表
val sinkDDL =
"CREATE TABLE test_sink (\n" +
" ip STRING,\n" +
" countSum BIGINT,\n" +
" PRIMARY KEY (ip) NOT ENFORCED\n" +
") WITH (\n" +
" 'connector' = 'print'\n" +
")"
val exeSQL =
"INSERT INTO test_sink " +
"SELECT ip, COUNT(1) " +
"FROM test_binlog " +
"GROUP BY ip"
tableEnv.executeSql(sourceDDL)
tableEnv.executeSql(sinkDDL)
val result = tableEnv.executeSql(exeSQL)
result.print()
}
INSERT INTO `test`.`t_test`( `ip`, `size`) VALUES (UUID(), 1231231);
INSERT INTO `test`.`t_test`( `ip`, `size`) VALUES (UUID(), 1231231);
INSERT INTO `test`.`t_test`( `ip`, `size`) VALUES (UUID(), 1231231);
...
插入數(shù)據(jù)可直接在console中看到flink處理的結(jié)果
Apache Flink CDC的方式替代了之前的canal+kafka節(jié)點.直接通過sql的方式來實現(xiàn)對mysql數(shù)據(jù)的同步。
看完上述內(nèi)容,你們掌握如何理解Apache Flink CDC原理與使用的方法了嗎?如果還想學到更多技能或想了解更多相關(guān)內(nèi)容,歡迎關(guān)注億速云行業(yè)資訊頻道,感謝各位的閱讀!
免責聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進行舉報,并提供相關(guān)證據(jù),一經(jīng)查實,將立刻刪除涉嫌侵權(quán)內(nèi)容。