Apache Flink CDC简介与使用

微信扫一扫,分享到朋友圈

Apache Flink CDC简介与使用

CDC (Change Data Capture)

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

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

  1. mysql开启binlog

  2. canal同步binlog数据写入到kafka

  3. 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的。

  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;
  1. 添加mysql-cdc相关依赖

<dependency>
<groupId>com.alibaba.ververica</groupId>
<artifactId>flink-connector-mysql-cdc</artifactId>
<version>1.1.0</version>
<scope>compile</scope>
</dependency>
  1. 相关代码实现

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()
}
  1. 启动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处理的结果

总结

Apache Flink CDC的方式替代了之前的canal+kafka节点.直接通过sql的方式来实现对mysql数据的同步。相关的完整代码我已提交至h ttps://github.com/CainGao/flink_learn   感兴趣的可以直接下来进行测试执行。

百亿级实时计算系统性能优化–—Elasticsearch篇

上一篇

潜水员深海发现神秘海洋生物:全身透明还不怕人

下一篇

你也可能喜欢

Apache Flink CDC简介与使用

长按储存图像,分享给朋友