FlinkCDC-自定义序列化器

package com.lcy.app.customer;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.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 CustomerBinlogDeserialization implements DebeziumDeserializationSchema<String> {
    /**
     * BINLOG格式：SourceRecord{
     *      sourcePartition={server=mysql_binlog_source},
     *      sourceOffset={file=mysql-bin.000003, pos=154, row=1, snapshot=true}}
     *      ConnectRecord{topic='mysql_binlog_source.gmall-210325-flink.user_info', kafkaPartition=null, key=Struct{id=3982},
     *      keySchema=Schema{mysql_binlog_source.gmall_210325_flink.user_info.Key:STRUCT},
     *      value=Struct{after=Struct{id=3982,login_name=g959roj95n,nick_name=冠策,name=司空冠策,phone_num=13339656498,email=g959roj95n@msn.com,user_level=2,birthday=1982-12-04,gender=M,create_time=2020-12-04 23:28:45},source=Struct{version=1.4.1.Final,connector=mysql,name=mysql_binlog_source,ts_ms=0,snapshot=true,db=gmall-210325-flink,table=user_info,server_id=0,file=mysql-bin.000003,pos=154,row=0},op=c,ts_ms=1639523578556},
     *      valueSchema=Schema{mysql_binlog_source.gmall_210325_flink.user_info.Envelope:STRUCT},
     *      timestamp=null,
     *      headers=ConnectHeaders(headers=)
     * }
     * @param sourceRecord
     * @param collector
     * @throws Exception
     */
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
        String topic = sourceRecord.topic();
        String[] tableInfos = topic.split("\\.");
        String tableName = tableInfos[2];
        String dbName = tableInfos[1];

        Struct value = (Struct)sourceRecord.value();
        Struct before = value.getStruct("before");
        List<Field> fields = null;
        JSONObject beforeJson = new JSONObject();
        if (before == null) {
            fields = before.schema().fields();
            fields.forEach(field -> {
                Object v = before.get(field.name());
                beforeJson.put(field.name(), v);
            });
        }

        Struct after = value.getStruct("after");
        JSONObject afterJson = new JSONObject();
        if (after != null) {
            fields = after.schema().fields();
            fields.forEach(field -> {
                Object v = after.get(field.name());
                afterJson.put(field.name(), v);
            });
        }

        Envelope.Operation operation = Envelope.operationFor(sourceRecord);
        String type = operation.toString().toLowerCase();
        if ("create".equals(type)) {
            type = "insert";
        }

        JSONObject result = new JSONObject();
        result.put("dbName",dbName);
        result.put("tableName",tableName);
        result.put("type",type);
        result.put("before",beforeJson);
        result.put("after",afterJson);

        collector.collect(result.toJSONString());
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }
}