Apache Flink CDC简介与使用

本文主要是介绍Apache Flink CDC简介与使用，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

CDC (Change Data Capture)

Flink在1.11版本中新增了CDC的特性，简称改变数据捕获。名称来看有点乱，我们先从之前的数据架构来看CDC的内容。

以上是之前的mysql binlog日志处理流程,例如canal监听binlog把日志写入到kafka中。而Apache Flink实时消费Kakfa的数据实现mysql数据的同步或其他内容等。拆分来说整体上可以分为以下几个阶段。

mysql开启binlog
canal同步binlog数据写入到kafka
flink读取kakfa中的binlog数据进行相关的业务处理。

整体的处理链路较长，需要用到的组件也比较多。Apache Flink CDC可以直接从数据库获取到binlog供下游进行业务计算分析。简单来说链路会变成这样

也就是说数据不再通过canal与kafka进行同步，而flink直接进行处理mysql的数据。节省了canal与kafka的过程。

Flink 1.11中实现了mysql-cdc与postgre-CDC，也就是说在Flink 1.11中我们可以直接通过Flink来直接消费mysql,postgresql的数据进行业务的处理。

使用场景

数据库数据的增量同步
数据库表之上的物理化视图
维表join
其他业务处理
...

MySQL CDC 操作实践

首先需要保证mysql数据库开启了binlog。未开启请查阅相关资料进行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 NULLPRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=183 DEFAULT CHARSET=utf8mb4;

添加mysql-cdc相关依赖

<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.getExecutionEnvironmentval 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()}

启动flink job,并且插入数据

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);
...

插入数据可直接在console中看到flink处理的结果