如何理解Apache Flink CDC原理与使用,相信很多没有经验的人对此束手无策,为此本文总结了问题出现的原因和解决方法,通过这篇文章希望你能解决这个问题。
Flink在1.11版本中新增了CDC的特性,简称 改变数据捕获。名称来看有点乱,我们先从之前的数据架构来看CDC的内容。
以上是之前的mysql binlog日志处理流程,例如canal监听binlog把日志写入到kafka中。而Apache Flink实时消费Kakfa的数据实现mysql数据的同步或其他内容等。拆分来说整体上可以分为以下几个阶段。
整体的处理链路较长,需要用到的组件也比较多。Apache Flink CDC可以直接从数据库获取到binlog供下游进行业务计算分析。简单来说链路会变成这样也就是说数据不再通过canal与kafka进行同步,而flink直接进行处理mysql的数据。节省了canal与kafka的过程。
Flink 1.11中实现了mysql-cdc与postgre-CDC,也就是说在Flink 1.11中我们可以直接通过Flink来直接消费mysql,postgresql的数据进行业务的处理。
首先需要保证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 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);...
插入数据可直接在console中看到flink处理的结果
Apache Flink CDC的方式替代了之前的canal+kafka节点.直接通过sql的方式来实现对mysql数据的同步。
看完上述内容,你们掌握如何理解Apache Flink CDC原理与使用的方法了吗?如果还想学到更多技能或想了解更多相关内容,欢迎关注亿速云行业资讯频道,感谢各位的阅读!
亿速云「云服务器」,即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘,价格低至29元/月。点击查看>>
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。
原文链接:https://my.oschina.net/CainGao/blog/4719367