本篇文章给大家分享的是有关flink 1.11中的CDC是什么意思,小编觉得挺实用的,因此分享给大家学习,希望大家阅读完这篇文章后可以有所收获,话不多说,跟着小编一起来看看吧。
CDC,Change Data Capture,变更数据获取的简称,使用CDC我们可以从数据库中获取已提交的更改并将这些更改发送到下游,供下游使用。这些变更可以包括INSERT,DELETE,UPDATE等,
用户可以在以下的场景下使用CDC:
使用flink sql进行数据同步,可以将数据从一个数据同步到其他的地方,比如mysql、elasticsearch等。
可以在源数据库上实时的物化一个聚合视图
因为只是增量同步,所以可以实时的低延迟的同步数据
使用EventTime join 一个temporal表以便可以获取准确的结果
flink 1.11 将这些changelog提取并转化为table apa和sql,目前支持两种格式:Debezium和Canal,这就意味着源表不仅仅是append操作,而且还有upsert、delete操作。
接下来我们使用canal为例简单介绍下CDC的使用
canal 格式:
{ "data": [ { "id": "13", "username": "13", "password": "6BB4837EB74329105EE4568DDA7DC67ED2CA2AD9", "name": "Canal Manager V2" } ], "old": [ { "id": "13", "username": "13", "password": "6BB4837EB74329105EE4568DDA7DC67ED2CA2AD9", "name": "Canal Manager" } ], "database": "canal_manager", "es": 1568972368000, "id": 11, "isDdl": false, "mysqlType": {...}, "pkNames": [ "id" ], "sql": "", "sqlType": {...}, "table": "canal_user", "ts": 1568972369005, "type": "UPDATE" }
简单讲下几个核心的字段:
type : 描述操作的类型,包括‘UPDATE’, 'INSERT', 'DELETE'。
data : 代表操作的数据。如果为'INSERT',则表示行的内容;如果为'UPDATE',则表示行的更新后的状态;如果为'DELETE',则表示删除前的状态。
old :可选字段,如果存在,则表示更新之前的内容,如果不是update操作,则为 null。
完整的语义如下;
private String destination; // 对应canal的实例或者MQ的topic private String groupId; // 对应mq的group id private String database; // 数据库或schema private String table; // 表名 private List<String> pkNames; private Boolean isDdl; private String type; // 类型: INSERT UPDATE DELETE // binlog executeTime private Long es; // 执行耗时 // dml build timeStamp private Long ts; // 同步时间 private String sql; // 执行的sql, dml sql为空 private List<Map<String, Object>> data; // 数据列表 private List<Map<String, Object>> old; // 旧数据列表, 用于update, size和data的size一一对应
-- 定义的字段和data 里面的数据想匹配 CREATE TABLE my_table ( id BIGINT, name STRING, description STRING, weight DECIMAL(10, 2) ) WITH ( 'connector' = 'kafka', 'topic' = 'products_binlog', 'properties.bootstrap.servers' = 'localhost:9092', 'properties.group.id' = 'testGroup', 'canal-json.ignore-parse-errors'='true' -- 忽略解析错误,缺省值false );
canal 格式也是作为一种flink的格式,而且是source,所以也就是涉及到读取数据的时候进行反序列化,我们接下来就简单看看CanalJson的反序列化的实现。具体的实现类是CanalJsonDeserializationSchema。
我们看下这个最核心的反序列化方法:
@Override public void deserialize(byte[] message, Collector<RowData> out) throws IOException { try { //使用json反序列化器将message反序列化成RowData RowData row = jsonDeserializer.deserialize(message); //获取type字段,用于下面的判断 String type = row.getString(2).toString(); if (OP_INSERT.equals(type)) { // 如果操作类型是insert,则data数组表示的是要插入的数据,则循环遍历data,然后添加一个标识INSERT,构造RowData对象,发送下游。 ArrayData data = row.getArray(0); for (int i = 0; i < data.size(); i++) { RowData insert = data.getRow(i, fieldCount); insert.setRowKind(RowKind.INSERT); out.collect(insert); } } else if (OP_UPDATE.equals(type)) { // 如果是update操作,从data字段里获取更新后的数据、 ArrayData data = row.getArray(0); // old字段获取更新之前的数据 ArrayData old = row.getArray(1); for (int i = 0; i < data.size(); i++) { // the underlying JSON deserialization schema always produce GenericRowData. GenericRowData after = (GenericRowData) data.getRow(i, fieldCount); GenericRowData before = (GenericRowData) old.getRow(i, fieldCount); for (int f = 0; f < fieldCount; f++) { if (before.isNullAt(f)) { //如果old字段非空,则说明进行了数据的更新,如果old字段是null,则说明更新前后数据一样,这个时候把before的数据也设置成after的,也就是发送给下游的before和after数据一样。 before.setField(f, after.getField(f)); } } before.setRowKind(RowKind.UPDATE_BEFORE); after.setRowKind(RowKind.UPDATE_AFTER); //把更新前后的数据都发送下游 out.collect(before); out.collect(after); } } else if (OP_DELETE.equals(type)) { // 如果是删除操作,data字段里包含将要被删除的数据,把这些数据组织起来发送给下游 ArrayData data = row.getArray(0); for (int i = 0; i < data.size(); i++) { RowData insert = data.getRow(i, fieldCount); insert.setRowKind(RowKind.DELETE); out.collect(insert); } } else { if (!ignoreParseErrors) { throw new IOException(format( "Unknown \"type\" value \"%s\". The Canal JSON message is '%s'", type, new String(message))); } } } catch (Throwable t) { // a big try catch to protect the processing. if (!ignoreParseErrors) { throw new IOException(format( "Corrupt Canal JSON message '%s'.", new String(message)), t); } } }
以上就是flink 1.11中的CDC是什么意思,小编相信有部分知识点可能是我们日常工作会见到或用到的。希望你能通过这篇文章学到更多知识。更多详情敬请关注亿速云行业资讯频道。
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。