场景
有时会有这种需求：Flink CDC 读取到程序，发送给下游软件框架，如 KAFKA
提出问题
通过 Flink CDC2.0 连接MySQL，将表原有数据打印到控制台 可知，控制台打印的是 SourceRecord、ConnectRecord 对象的 toString() 结果，不利于程序读取数据，如下：
SourceRecord{
sourcePartition={server=mysql_binlog_source}, # mysql binlog
sourceOffset={
ts_sec=1655552829, # binlog 时间戳
file=mysql-bin.000004, # binlog 文件名
pos=154,
snapshot=true
}
}
ConnectRecord{
topic='mysql_binlog_source.scott.dept', # 表名
kafkaPartition=null,
key=Struct{deptno=7369}, # 主键
keySchema=Schema{mysql_binlog_source.scott.dept.Key:STRUCT},
value=Struct{
after=Struct{deptno=74,dname=财务部,loc=杭州2}, # 数据
source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1655552829388,snapshot=true,db=scott,table=dept,server_id=0,file=mysql-bin.000004,pos=154,row=0},
op=r, # crud中的 Retrieve，表示查询出的数据
ts_ms=1655552829392
}, valueSchema=Schema{mysql_binlog_source.scott.dept.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
解决问题
自定义 反序列化器，定制通用、易读取的 输出格式，如：JSON 格式
添加依赖
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.76</version>
</dependency>
java 代码
自定义反序列化器
JsonDebeziumDeserializationSchema，仿照 StringDebeziumDeserializationSchema ，实现 DebeziumDeserializationSchema 接口
代码如下：
package std2;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;
import java.util.List;
public class JsonDebeziumDeserializationSchema implements DebeziumDeserializationSchema<String> {
/**
* 获取数据库名、表名
* @param sourceRecord
* @return
*/
private JSONObject getDabaseTable(SourceRecord sourceRecord){
String topic = sourceRecord.topic();
String[] fields = topic.split("\\.");
String database = fields[1];
String tableName = fields[2];
JSONObject ret = new JSONObject();
ret.put("database", database);
ret.put("table", tableName);
return ret;
}
/**
* 获取before、after数据
* @param sourceRecord
* @param fieldName
* @return
*/
public JSONObject getDataJson(SourceRecord sourceRecord,String fieldName){
Struct value = (Struct) sourceRecord.value();
Struct struct = value.getStruct(fieldName);
JSONObject ret = new JSONObject();
if (struct != null) {
Schema schema = struct.schema();
List<Field> fields = schema.fields();
for (Field field : fields) {
Object obj = struct.get(field);
ret.put(field.name(), obj);
}
}
return ret;
}
public String getOP(SourceRecord sourceRecord){
Envelope.Operation operation = Envelope.operationFor(sourceRecord);
// String type = operation.toString().toLowerCase();
String type = operation.code();
return type;
}
@Override
public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
//1.创建 JSON 对象用于存储最终数据
JSONObject result = new JSONObject();
//2.获取库名、表名放入 source
JSONObject databaseTableJson = getDabaseTable(sourceRecord);
//3.获取"before"数据
JSONObject beforeJson = getDataJson(sourceRecord, "before");
//4.获取"after"数据
JSONObject afterJson = getDataJson(sourceRecord, "after");
//5.获取操作类型 CREATE UPDATE DELETE 进行符合 Debezium-op 的字母
String type = getOP(sourceRecord);
//6.将字段写入 JSON 对象
result.put("source", databaseTableJson);
result.put("before", beforeJson);
result.put("after", afterJson);
result.put("op", type);
//7.输出数据
collector.collect(result.toJSONString());
}
@Override
public TypeInformation<String> getProducedType() {
// 表示返回String类型
return BasicTypeInfo.STRING_TYPE_INFO;
}
}
启动类
代码如下：
public class S2 {
public static void main(String[] args) throws Exception {
DebeziumSourceFunction<String> build = MySqlSource.<String>builder()
.hostname("hadoop1")
.port(3306)
.databaseList("scott") // 设置要捕获数据的数据库，可传入多个数据库
.tableList("scott.dept") // 设置要捕获数据的表，必须是 【库名.表名】的形式。如果不传，表示监控上面数据库中所有表的数据
.username("root")
.password("root")
.deserializer(new JsonDebeziumDeserializationSchema()) // 指定自定义反序列化器
.startupOptions(StartupOptions.initial())
.build();
// 获取flink环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 设置并行度
env.setParallelism(1);
// 需要设置超时时间，否则可能会因为连接时间过长，报错连接失败
// env.enableCheckpointing(3000);
// 添加数据源
DataStreamSource<String> source = env.addSource(build);
// 打印
source.print();
env.execute("Print MySQL Snapshot + Binlog");
}
}
打印结果
启动
打印表中原有数据，如下：
{
"op":"r", # 操作，crud中的r
"before":{ # 是空的
},
"source":{
"database":"scott",
"table":"dept"
},
"after":{ # 表中原有数据
"loc":"北京",
"dname":"市场部",
"deptno":3
}
}
插入数据
执行下面SQL：
INSERT INTO `scott`.`dept`(`deptno`, `dname`, `loc`) VALUES (60, '测试部', 'NEW YORK');
打印结果如下：
{
"op":"c", # 操作，crud中的c
"before":{
},
"source":{
"database":"scott",
"table":"dept"
},
"after":{ # 插入后数据
"loc":"NEW YORK",
"dname":"测试部",
"deptno":60
}
}
修改数据
执行下面SQL：
update `scott`.`dept` set loc='伦敦' where deptno = 60
打印结果如下：
{
"op":"u", # 操作，crud中的u
"before":{ # 修改前数据
"loc":"杭州",
"dname":"测试部",
"deptno":60
},
"source":{
"database":"scott",
"table":"dept"
},
"after":{ # 修改后数据
"loc":"伦敦",
"dname":"测试部",
"deptno":60
}
}
删除数据
打印结果如下：
{
"op":"d", # 操作，crud中的d
"before":{ # 删除前数据
"loc":"伦敦",
"dname":"测试部",
"deptno":60
},
"source":{
"database":"scott",
"table":"dept"
},
"after":{
}
}