CDC (Change Data Capture)

Flink在1.11版本中新增了CDC的特性，簡稱 改變數(shù)據(jù)捕獲。名稱來看有點(diǎn)亂，我們先從之前的數(shù)據(jù)架構(gòu)來看CDC的內(nèi)容。

以上是之前的MySQL binlog日志處理流程,例如canal監(jiān)聽binlog把日志寫入到kafka中。而Apache Flink實(shí)時(shí)消費(fèi)Kakfa的數(shù)據(jù)實(shí)現(xiàn)mysql數(shù)據(jù)的同步或其他內(nèi)容等。拆分來說整體上可以分為以下幾個(gè)階段。

1. mysql開啟binlog
2. canal同步binlog數(shù)據(jù)寫入到kafka
3. flink讀取kakfa中的binlog數(shù)據(jù)進(jìn)行相關(guān)的業(yè)務(wù)處理。

整體的處理鏈路較長，需要用到的組件也比較多。Apache Flink CDC可以直接從數(shù)據(jù)庫獲取到binlog供下游進(jìn)行業(yè)務(wù)計(jì)算分析。簡單來說鏈路會(huì)變成這樣也就是說數(shù)據(jù)不再通過canal與kafka進(jìn)行同步，而flink直接進(jìn)行處理mysql的數(shù)據(jù)。節(jié)省了canal與kafka的過程。

Flink 1.11中實(shí)現(xiàn)了mysql-cdc與postgre-CDC，也就是說在Flink 1.11中我們可以直接通過Flink來直接消費(fèi)mysql,postgresql的數(shù)據(jù)進(jìn)行業(yè)務(wù)的處理。

使用場(chǎng)景

• 數(shù)據(jù)庫數(shù)據(jù)的增量同步
• 數(shù)據(jù)庫表之上的物理化視圖
• 維表join
• 其他業(yè)務(wù)處理
• ...

MySQL CDC 操作實(shí)踐

首先需要保證mysql數(shù)據(jù)庫開啟了binlog。未開啟請(qǐng)查閱相關(guān)資料進(jìn)行binlog的啟用。自建默認(rèn)是不開啟binlog的。

1. 源表

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;

2. 添加mysql-cdc相關(guān)依賴

  com.alibaba.ververica
  flink-connector-mysql-cdc
  1.1.0
  compile

3. 相關(guān)代碼實(shí)現(xiàn)

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'" +
      ")"

    // 輸出目標(biāo)表
    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()
  }

4. 啟動(dòng)flink job,并且插入數(shù)據(jù)

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é)點(diǎn).直接通過sql的方式來實(shí)現(xiàn)對(duì)mysql數(shù)據(jù)的同步。