总览
变更数据捕获 (CDC) 已成为一种流行的模式，用于从数据库捕获已提交的变更并将这些变更传播给下游消费者，例如保持多个数据存储同步并避免常见的陷阱，例如双重写入。 能够轻松地将这些变更日志摄取和解释到 Table API/SQL 中一直是 Flink 社区的一个非常需要的功能，现在 Flink 1.11 可以实现。
为了将 Table API/SQL 的范围扩展到 CDC 等用例，Flink 1.11 引入了具有变更日志模式的新表源和接收器接口（请参阅新的 TableSource 和 TableSink 接口）并支持 Debezium 和 Canal 格式（FLIP-105）。 这意味着动态表源不再仅限于追加操作，并且可以摄取这些外部更改日志（插入事件），将它们解释为更改操作（插入、更新、删除事件）并使用更改类型将它们发送到下游。
用户必须在其 CREATE TABLE 语句中指定“format=debezium-json”或“format=canal-json”才能使用 SQL DDL 使用更改日志。
例如：
CREATE TABLE my_table (
...
) WITH (
'connector'='...', -- e.g. 'kafka'
'format'='debezium-json',
'debezium-json.schema-include'='true' -- default: false (Debezium can be configured to include or exclude the message schema)
'debezium-json.ignore-parse-errors'='true' -- default: false
);目前，Flink内置Debezium CDC、Canal CDC、Maxwell CDC三种format并以kafka和Filesystem作为connector。但是，我们在使用的时候发现，其实上述三种CDC format是远远不能满足我们的需求的公司客户有各种各样的自定义CDC格式。下面列举其中一种格式，并针对此格式自定义CDC format。
有什么参考资料？
参考官方CDC format用例，具体为flink-json-1.1x.x.jar包中提供的内容。
如
org.apache.flink.formats.json.canal
org.apache.flink.formats.json.debezium
org.apache.flink.formats.json.maxwell
本文自定义CDC数据源格式
只有I和D格式，U已经被拆解为D和I
I格式数据：
{
"columns": [
"col1",
"col2",
"col3",
"col4",
"col5"
],
"rows": [
{
"after": [
"a1",
"b1",
"c1",
"d1",
"e1"
],
"current_ts": "2021-09-23 17:11:33",
"op_ts": "2021-09-23 17:11:33",
"op_type": "I",
"pos": "09231632388293978189"
},
{
"after": [
"a2",
"b2",
"c2",
"d2",
"e2"
],
"current_ts": "2021-09-23 17:11:33",
"op_ts": "2021-09-23 17:11:33",
"op_type": "I",
"pos": "09231632388293978189"
}
],
"table": "table1"
}D格式数据
{
"columns": [
"col1",
"col2",
"col3",
"col4",
"col5"
],
"rows": [
{
"before": [
"a1",
"b1",
"c1",
"d1",
"e1"
],
"current_ts": "2021-09-23 17:11:33",
"op_ts": "2021-09-23 17:11:33",
"op_type": "D",
"pos": "09231632388293978189"
},
{
"before": [
"a2",
"b2",
"c2",
"d2",
"e2"
],
"current_ts": "2021-09-23 17:11:33",
"op_ts": "2021-09-23 17:11:33",
"op_type": "D",
"pos": "09231632388293978189"
}
],
"table": "table1"
}数据说明：为了减少传输带宽，如果是批量D或者I的时候，在一条数据中传输多个相同类型的数据。比如说上述I操作，columns表示字段名称，在rows中定义多条插入数据。由于U被拆分为，D和I，数据传输为单条数据传输，也就是rows中不会出现多条数据的情况，并且确保顺序写入kafka同一分区。
从何写起
下图描述了Maxwell CDC format相关流程：
![[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-g4r7l3XK-1634882436700)(https://obs-githubhelper.obs.cn-east-3.myhuaweicloud.com/blog-images/category/bigdata/flink/flink-cdc-user-define-format/image%20(1)].png)](https://ucc.alicdn.com/images/user-upload-01/0b2cb7394d5e43b993594bae8c6e8181.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAQmlnRGF0YVRvQUk=,size_20,color_FFFFFF,t_70,g_se,x_16&x-oss-process=image/resize,w_1400/format,webp "[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-g4r7l3XK-1634882436700)(https://obs-githubhelper.obs.cn-east-3.myhuaweicloud.com/blog-images/category/bigdata/flink/flink-cdc-user-define-format/image%20(1)].png)")
从上图我们可以知道，Flink 会通过 SPI 机制将 classpath 下注册的所有工厂类加载进来，包括 DynamicTableFactory、DeserializationFormatFactory 等等。同时通过DDL中的format的值与DeserializationFormatFactory工厂类的factoryIdentifier()返回值进行匹配，从而确定使用哪个工厂类。
然后再通过 DeserializationFormatFactory 中的 createDecodingFormat(...) 方法，将反序列化对象提供给 DynamicTableSource。
所以说，上述流程主要完成了以下工作：
- 定义工厂类（DeserializationFormatFactory），即 MaxwellJsonFormatFactory，负责编译时根据 ‘format’ = ‘maxwell-json’创建对应的反序列化器
- 定义反序列化类（DeserializationSchema），即MaxwellJsonDeserializationSchema，负责运行时的解析，根据固定格式将 CDC 数据转换成 Flink 系统能认识的 INSERT/DELETE/UPDATE 消息，如 RowData。
- Service 注册文件：需要添加 Service 文件 META-INF/services/org.apache.flink.table.factories.Factory ，并在其中增加一行上述实现的 MaxwellJsonFormatFactory 类路径。
开始定义你的cdc format
根据上述Maxwell CDC format相关流程，我们先定义工厂类，具体如下：
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.formats.common.TimestampFormat;
import org.apache.flink.formats.json.JsonOptions;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.format.DecodingFormat;
import org.apache.flink.table.connector.format.EncodingFormat;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.factories.DeserializationFormatFactory;
import org.apache.flink.table.factories.DynamicTableFactory;
import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.table.factories.SerializationFormatFactory;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.types.RowKind;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
public class AnalysisJsonFormatFactory implements DeserializationFormatFactory, SerializationFormatFactory {
public static final String IDENTIFIER = "analysis-json";
public AnalysisJsonFormatFactory() {
}
@Override
public DecodingFormat<DeserializationSchema<RowData>> createDecodingFormat(DynamicTableFactory.Context context, ReadableConfig formatOptions) {
FactoryUtil.validateFactoryOptions(this, formatOptions);
AnalysisJsonOptions.validateDecodingFormatOptions(formatOptions);
boolean ignoreParseErrors = (Boolean)formatOptions.get(AnalysisJsonOptions.IGNORE_PARSE_ERRORS);
TimestampFormat timestampFormat = JsonOptions.getTimestampFormat(formatOptions);
return new DecodingFormat<DeserializationSchema<RowData>>() {
@Override
public DeserializationSchema<RowData> createRuntimeDecoder(DynamicTableSource.Context context, DataType dataType) {
// 表的字段名和数据类型
RowType rowType = (RowType) dataType.getLogicalType();
TypeInformation<RowData> rowDataTypeInformation = context.createTypeInformation(dataType);
return new AnalysisJsonDeserializationSchema(rowType, rowDataTypeInformation, ignoreParseErrors, timestampFormat);
}
@Override
public ChangelogMode getChangelogMode() {
return ChangelogMode.newBuilder().addContainedKind(RowKind.INSERT).addContainedKind(RowKind.DELETE).build();
}
};
}
@Override
public EncodingFormat<SerializationSchema<RowData>> createEncodingFormat(DynamicTableFactory.Context context, ReadableConfig readableConfig) {
return null;
}
@Override
public String factoryIdentifier() {
return "analysis-json";
}
@Override
public Set<ConfigOption<?>> requiredOptions() {
return Collections.emptySet();
}
@Override
public Set<ConfigOption<?>> optionalOptions() {
Set<ConfigOption<?>> options = new HashSet<>();
options.add(AnalysisJsonOptions.IGNORE_PARSE_ERRORS);
return options;
}
}定义反序列化器：
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.formats.common.TimestampFormat;
import org.apache.flink.formats.json.JsonRowDataDeserializationSchema;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.types.RowKind;
import org.apache.flink.util.Collector;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
public class AnalysisJsonDeserializationSchema implements DeserializationSchema<RowData> {
private static final long serialVersionUID = 1L;
private static final String OP_INSERT = "I";
private static final String OP_DELETE = "D";
private final JsonRowDataDeserializationSchema jsonDeserializer;
private final TypeInformation<RowData> resultTypeInfo;
private final boolean ignoreParseErrors;
private final int fieldCount;
public AnalysisJsonDeserializationSchema(RowType rowType, TypeInformation<RowData> resultTypeInfo, boolean ignoreParseErrors, TimestampFormat timestampFormatOption) {
this.resultTypeInfo = resultTypeInfo;
this.ignoreParseErrors = ignoreParseErrors;
this.fieldCount = rowType.getFieldCount();
this.jsonDeserializer = new JsonRowDataDeserializationSchema(this.createJsonRowType(), resultTypeInfo, false, ignoreParseErrors, timestampFormatOption);
}
@Override
public RowData deserialize(byte[] bytes) throws IOException {
throw new RuntimeException("Please invoke DeserializationSchema#deserialize(byte[], Collector<RowData>) instead.");
}
public void deserialize(byte[] bytes, Collector<RowData> out) throws IOException {
if (bytes != null && bytes.length != 0) {
RowData rowData = this.jsonDeserializer.deserialize(bytes);
String columns = rowData.getString(0).toString();
JSONArray columnsJSONArray = JSONArray.parseArray(columns);
String rows = rowData.getString(1).toString();
JSONArray rowsJSONArray = JSONArray.parseArray(rows);
for (Object object : rowsJSONArray) {
JSONObject rowJSONObject = (JSONObject) object;
Map<String, String> outputData = new HashMap<>();
if (OP_INSERT.equals(rowJSONObject.getString("op_type"))) {
JSONArray after = rowJSONObject.getJSONArray("after");
int index = 0;
for (Object column : columnsJSONArray) {
outputData.put(column.toString(), after.getString(index));
index++;
}
GenericRowData insert = new GenericRowData(1);
insert.setField(0, StringData.fromBytes(JSONObject.toJSONBytes(outputData)));
insert.setRowKind(RowKind.INSERT);
out.collect(insert);
}
else {
JSONArray before = rowJSONObject.getJSONArray("before");
int index = 0;
for (Object column : columnsJSONArray) {
outputData.put(column.toString(), before.getString(index));
index++;
}
GenericRowData delete = new GenericRowData(1);
delete.setField(0, StringData.fromBytes(JSONObject.toJSONBytes(outputData)));
delete.setRowKind(RowKind.DELETE);
out.collect(delete);
}
}
}
}
@Override
public boolean isEndOfStream(RowData rowData) {
return false;
}
@Override
public TypeInformation<RowData> getProducedType() {
return this.resultTypeInfo;
}
public boolean equals(Object o) {
if (this == o) {
return true;
} else if (o != null && this.getClass() == o.getClass()) {
AnalysisJsonDeserializationSchema that = (AnalysisJsonDeserializationSchema)o;
return this.ignoreParseErrors == that.ignoreParseErrors && this.fieldCount == that.fieldCount && Objects.equals(this.jsonDeserializer, that.jsonDeserializer) && Objects.equals(this.resultTypeInfo, that.resultTypeInfo);
} else {
return false;
}
}
public int hashCode() {
return Objects.hash(this.jsonDeserializer, this.resultTypeInfo, this.ignoreParseErrors, this.fieldCount);
}
private RowType createJsonRowType() {
return (RowType) DataTypes.ROW(
DataTypes.FIELD("columns", DataTypes.STRING()),
DataTypes.FIELD("rows", DataTypes.STRING()),
DataTypes.FIELD("table", DataTypes.STRING())).getLogicalType();
}
}具体来看一下上述代码，工厂类AnalysisJsonFormatFactory的写法相对固定，在createDecodingFormat中返回DecodingFormat，并重写createRuntimeDecoder和getChangelogMode方法，其中，createRuntimeDecoder返回的为上述定义的AnalysisJsonDeserializationSchema，getChangelogMode则返回我们数据格式中可能存在的操作形式，本位为INSERT以及DELETE。
再来看一下AnalysisJsonDeserializationSchema，其中this.jsonDeserializer则描述了如何反序列化原始kafka数据，在本例中，由于原始数据格式固定，所以直接定义其格式为：
private RowType createJsonRowType() {
return (RowType) DataTypes.ROW(
DataTypes.FIELD("columns", DataTypes.STRING()),
DataTypes.FIELD("rows", DataTypes.STRING()),
DataTypes.FIELD("table", DataTypes.STRING())).getLogicalType();
}再来看一下核心方法，deserialize(byte[] bytes, Collector out)，该方法接受原始byte[]类型原始数据，并通过out返回解析完的数据。
该方法通过this.jsonDeserializer将原始数据反序列化为rowData，那么后续则可以通过此rowData获取原始数据中的columns、rows以及table中的值。
上述代码中解析rows中的数据，并判断其类型为I或者D，并封装为GenericRowData返回。
在上一个步骤中，我们说到要注册工厂类，所以我们需要在项目resources目录中新建META-INF/services/org.apache.flink.table.factories.Factory，并在org.apache.flink.table.factories.Factory文件中填入：AnalysisJsonFormatFactory工厂类的完整路径。
完整代码示例
https://github.com/xiaozhch5/flink-cdc-format.git
参考资料
https://lrting.top/special/hudi/
