赞
踩
未来总是藏在迷雾中让人胆怯,但当你踏入其中,便会云开雾散
CDC是Change Data Capture(变更数据获取)的简称。核心思想是,检测并捕获数据库的变动(包括数据或数据表的插入、更新以及删除等),将这些变更按发生的顺序完整记录下来,写入到消息中间件以及其他服务进行订阅及消费。
CDC主要分为基于查询和基于Binlog两种方式,以下这两种之间的区别:
基于查询的CDC | 基于Binlog的CDC | |
---|---|---|
开源产品 | Sqoop、DataX | Canal、Maxwell |
执行模式 | Batch | Streaming |
是否可以捕获所有数据变化 | 否 | 是 |
延迟性 | 高延迟 | 低延迟 |
是否增加数据库压力 | 是 | 否 |
Flink摄取开发了Flink-cdc-connectors组件,这是一个可以直接从MySQL、PostgreSQL等数据库直接读取全量数据和增量变更数据的Source组件,
目前已经开源,开源地址:https://github.com/ververica/flink-cdc-connectors
vim /etc/my.cnf
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-planner-blink-loader
</artifactId>
<version>1.17.0</version>
</dependency>
public class FlinkCDC_SQL { public static void main(String[] args) throws Exception { // TODO 1. 准备环境 // 1.1 流处理环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 1.2 表执行环境 StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env); // TODO 2. 创建动态表 tableEnv.executeSql("CREATE TABLE user_info (\n" + "id INT,\n" + "name STRING,\n" + "age INT,\n" + "primary key(id) not enforced\n" + ") WITH (" + "'connector' = 'mysql-cdc'," + "'hostname' = 'hadoop102'," + "'port' = '3306'," + "'username' = 'root'," + "'password' = '123456'," + "'database-name' = 'test'," + "'table-name' = 'user'" + ")"); tableEnv.executeSql("select * from user_info").print(); // TODO 3. 执行任务 env.execute(); } }
运行截图:
MySQL表数据截图:
您的支持是我创作的无限动力
希望我能为您的未来尽绵薄之力
如有错误,谢谢指正;若有收获,谢谢赞美
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。