说明
通过本文案例，执行 增删改 SQL语句，查看 Flink CDC 打印数据的格式
准备
Flink CDC2.0 连接MySQL，将表原有数据打印到控制台
插入数据
一条语句插入一条记录
执行下面sql：
INSERT INTO `scott`.`dept`(`deptno`, `dname`, `loc`) VALUES (60, '测试部', 'NEW YORK');
输出内容如下：
SourceRecord{
sourcePartition={server=mysql_binlog_source}, # 从 binlog 读取数据
sourceOffset={transaction_id=null, ts_sec=1655560638, file=mysql-bin.000004, pos=2185, row=1, server_id=1, event=2}}
ConnectRecord{topic='mysql_binlog_source.scott.dept', # 数据库、表
kafkaPartition=null, key=Struct{deptno=60}, keySchema=Schema{mysql_binlog_source.scott.dept.Key:STRUCT},value=Struct{after=Struct{deptno=60,dname=测试部,loc=NEW YORK},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1655560638000,db=scott,table=dept,server_id=1,file=mysql-bin.000004,pos=2307,row=0},
op=c, # crud中的c，表示添加一条记录
ts_ms=1655560638700}, valueSchema=Schema{mysql_binlog_source.scott.dept.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
一条语句插入多条记录
执行下面sql：
INSERT INTO `scott`.`dept`(`deptno`, `dname`, `loc`) VALUES (71, '财务部', '上海'),(72,'HR','北京'),(73,'销售部','广州');
结论： 一条sql插入多条记录，在Flink CDC中，仍然是3条记录
修改数据
执行下面sql：
update `scott`.`dept` set loc='杭州' where deptno = 60
打印结果如下：
SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={transaction_id=null, ts_sec=1655560980, file=mysql-bin.000004, pos=2462, row=1, server_id=1, event=2}} ConnectRecord{topic='mysql_binlog_source.scott.dept', kafkaPartition=null, key=Struct{deptno=60}, keySchema=Schema{mysql_binlog_source.scott.dept.Key:STRUCT}, value=Struct{
before=Struct{deptno=60,dname=测试部,loc=NEW YORK}, # 修改前数据
after=Struct{deptno=60,dname=测试部,loc=杭州}, # 修改后数据
source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1655560980000,db=scott,table=dept,server_id=1,file=mysql-bin.000004,pos=2584,row=0},
op=u, # crud 中的 u，表示更新操作
ts_ms=1655560980866}, valueSchema=Schema{mysql_binlog_source.scott.dept.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
删除数据
delete from dept where deptno = 73
maxwell打印结果如下：
SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={transaction_id=null, ts_sec=1655560464, file=mysql-bin.000004, pos=494, row=1, server_id=1, event=2}} ConnectRecord{topic='mysql_binlog_source.scott.dept', kafkaPartition=null, key=Struct{deptno=75}, keySchema=Schema{mysql_binlog_source.scott.dept.Key:STRUCT}, value=Struct{
before=Struct{deptno=75,dname=财务部,loc=上海}, # 之前的数据
source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1655560464000,db=scott,table=dept,server_id=1,file=mysql-bin.000004,pos=616,row=0},
op=d, # crud中的d，表示删除操作
ts_ms=1655560464620}, valueSchema=Schema{mysql_binlog_source.scott.dept.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}