cdc : change data capture 变更数据捕获
它可以监听数据库的数据变化,并将变化数据进行记录
pom
org.apache.flink flink-streaming-java_2.11 1.12.0 org.apache.flink flink-clients_2.11 1.12.0 mysql mysql-connector-java 8.0.16 com.alibaba.ververica flink-connector-mysql-cdc 1.0.0 org.slf4j slf4j-api 1.7.25 org.slf4j slf4j-log4j12 1.7.25
java代码
package com.antg.flink_cdc;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
public class Demo01 {
public static void main(String[] args) throws Exception {
//创建监听的数据源
SourceFunction sourceFunction = MySQLSource.builder()
.hostname("localhost")
.port(3306)
.databaseList("test") // 监听test数据库下的所有表
.username("root")
.password("12345678")
.deserializer(new StringDebeziumDeserializationSchema()) // 将SourceRecord 转换成 String形式
.build();
//获取运行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//开始监听数据,并将动态数据变化打印到屏幕中
env.addSource(sourceFunction).print();
//开始执行任务
env.execute();
}
}
测试
测试sql
-- op=c 添加
insert into user(name) values("123");
-- op=d 删除
delete from user where id=1;
-- op=u 更新
update user set name="fjh" where id=2;
监听结果
添加
SourceRecord{sourcePartition={server=mysql-binlog-source}, sourceOffset={ts_sec=1641896364, file=binlog.000018, pos=2237, row=1, server_id=1, event=2}} ConnectRecord{topic='mysql-binlog-source.test.user', kafkaPartition=null, key=Struct{id=33}, keySchema=Schema{mysql_binlog_source.test.user.Key:STRUCT}, value=Struct{after=Struct{id=33,name=123},source=Struct{version=1.2.0.Final,connector=mysql,name=mysql-binlog-source,ts_ms=1641896364000,db=test,table=user,server_id=1,file=binlog.000018,pos=2368,row=0,thread=8},op=c,ts_ms=1641896364267}, valueSchema=Schema{mysql_binlog_source.test.user.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
删除
SourceRecord{sourcePartition={server=mysql-binlog-source}, sourceOffset={ts_sec=1641896499, file=binlog.000018, pos=3095, row=1, server_id=1, event=2}} ConnectRecord{topic='mysql-binlog-source.test.user', kafkaPartition=null, key=Struct{id=1}, keySchema=Schema{mysql_binlog_source.test.user.Key:STRUCT}, value=Struct{before=Struct{id=1,name=tom},source=Struct{version=1.2.0.Final,connector=mysql,name=mysql-binlog-source,ts_ms=1641896499000,db=test,table=user,server_id=1,file=binlog.000018,pos=3226,row=0,thread=8},op=d,ts_ms=1641896499530}, valueSchema=Schema{mysql_binlog_source.test.user.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
更新
SourceRecord{sourcePartition={server=mysql-binlog-source}, sourceOffset={ts_sec=1641897004, file=binlog.000018, pos=3381, row=1, server_id=1, event=2}} ConnectRecord{topic='mysql-binlog-source.test.user', kafkaPartition=null, key=Struct{id=2}, keySchema=Schema{mysql_binlog_source.test.user.Key:STRUCT}, value=Struct{before=Struct{id=2,name=13:22:40},after=Struct{id=2,name=fjh},source=Struct{version=1.2.0.Final,connector=mysql,name=mysql-binlog-source,ts_ms=1641897004000,db=test,table=user,server_id=1,file=binlog.000018,pos=3521,row=0,thread=8},op=u,ts_ms=1641897004887}, valueSchema=Schema{mysql_binlog_source.test.user.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
注意 : 在使用之前一定要开始mysql的binlog,因为flinkced是基于binlog做数据监听的
三.监听结果详解其中用于区别监听类型的key是op,监听结果中包含了如下几部分
为了方便显示,我们将其中一条进行格式化一下
和json的格式比较像
SourceRecord {
sourcePartition = {
server = mysql - binlog - source},
sourceOffset = {
ts_sec = 1641896364,
file = binlog .000018,
pos = 2237,
row = 1,
server_id = 1,
event = 2
}
}
ConnectRecord {
topic = 'mysql-binlog-source.test.user', kafkaPartition = null, key = Struct {
id = 33
}, keySchema = Schema {
mysql_binlog_source.test.user.Key: STRUCT
}, value = Struct {
after = Struct {
id = 33, name = 123
}, source = Struct {
version = 1.2 .0.Final, connector = mysql, name = mysql - binlog - source, ts_ms = 1641896364000, db = test, table = user, server_id = 1, file = binlog .000018, pos = 2368, row = 0, thread = 8
}, op = c, ts_ms = 1641896364267
}, valueSchema = Schema {
mysql_binlog_source.test.user.Envelope: STRUCT
}, timestamp = null, headers = ConnectHeaders(headers = )
}
主要分为两个部分:SourceRecord和ConnectRecord,分别记录了数据源信息和连接信息
其中最重要的数据便是value字段,它记录了那些数据变化了
其中字段op,就是操作类型
op=c 添加
op=d 删除
op=u 更新
