本篇文章给大家分享的是有关flink 1.11中的CDC是什么意思，小编觉得挺实用的，因此分享给大家学习，希望大家阅读完这篇文章后可以有所收获，话不多说，跟着小编一起来看看吧。
成都创新互联公司专注为客户提供全方位的互联网综合服务，包含不限于网站设计、成都网站建设、济水街道网络推广、小程序定制开发、济水街道网络营销、济水街道企业策划、济水街道品牌公关、搜索引擎seo、人物专访、企业宣传片、企业代运营等，从售前售中售后，我们都将竭诚为您服务，您的肯定，是我们最大的嘉奖；成都创新互联公司为所有大学生创业者提供济水街道建站搭建服务，24小时服务热线：13518219792，官方网址：www.cdcxhl.com
CDC简介
CDC,Change Data Capture,变更数据获取的简称，使用CDC我们可以从数据库中获取已提交的更改并将这些更改发送到下游，供下游使用。这些变更可以包括INSERT,DELETE,UPDATE等，
用户可以在以下的场景下使用CDC：
flink 1.11 将这些changelog提取并转化为table apa和sql，目前支持两种格式：Debezium和Canal，这就意味着源表不仅仅是append操作，而且还有upsert、delete操作。
Canal
接下来我们使用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 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> data; // 数据列表
private List> 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
);
CanalJson反序列化源码解析
canal 格式也是作为一种flink的格式，而且是source，所以也就是涉及到读取数据的时候进行反序列化，我们接下来就简单看看CanalJson的反序列化的实现。具体的实现类是CanalJsonDeserializationSchema。
我们看下这个最核心的反序列化方法：
@Override
public void deserialize(byte[] message, Collector 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是什么意思，小编相信有部分知识点可能是我们日常工作会见到或用到的。希望你能通过这篇文章学到更多知识。更多详情敬请关注创新互联行业资讯频道。
标题名称：flink1.11中的CDC是什么意思
URL地址：
http://bzwzjz.com/article/gpohhd.html
