cdc , 流水理鱼

1. 需求

业务上对业务表就行重构，重新设计了新表，需要原有的数据准实时同步到新表（目标表）

字段类型和数据值等做了修改需要处理映射

例如

1是，2否需要映射为 1是，0否

varchar存储的时间转换为datetime

int存储的时间戳要转换为datetime

新表设计了新状态码，需要按照对照表做数据值映射

三表合一逻辑结构为

源表主键/唯一建	类别	需要的字段	源表	目标表	目标表主键
id	事实主表	10+	old_orders	new_orders	id（订单号）
order_id	业务延迟产生数据	2+	old_orders_extend1
order_id	业务延迟产生数据	1个	old_orders_extend2

需要实现，三张源表的的insert和update操作准实时的方式同步到同目标目标表中

2.设计

2.1 功能性设计

准实时，必须使用MySQL binlog订阅CDC的方式，使用CloudCanal作为准实时同步数据平台

字段类型和数据值的转化映射，使用CloudCanal的自定义数据处理插件功能，自编写业务处理代码（java）实现

2.2 鲁棒性设计

建表

null处理，目标表字段设置为允许为null，虽然在数据库上常规来讲设置为not null更性能，但是考虑综合收益新表就尽量不使用 not null，这样有更好的数据导入容错性

容量，目标表表字段容量大小必须>= 源表对应的字段，一般情况保持原有大小一致

数据转换

格式化，时间字段格式化，文本格式的字段类型转换为datatime类型时做好异常捕获，确保格式化为正确的日期，保障数据写入更新不出问题

脏数据，其他脏数据的预处理

2.3 CloudCanal任务设计

事实主表建立一个任务实例，这样做保障事实主表独立

业务延迟产生数据的表建立一个任务实例，这样做和事实主表的同步独立，互不影响

3.实现

3.1 目标表建表

注意符合鲁棒性设计中的建表设计

具体内容略…

3.2 编写自定义数据处理插件

常规的一些转换操作看CloudCanal提供的示例代码即可

处理技术重点在，业务延迟产生数据的时候，需要把insert和update都转换为update

处理insert时候，需要同时保留getAfterColumnMap和getBeforeColumnMap，由于insert是没有变更前数据的，所以设置getBeforeColumnMap为getAfterColumnMap即可

核心代码为

//  本文来自 www.iamle.com 流水理鱼
public static void changeOrderExtend1FillColumnMap(CustomFieldV2 oriBf, LinkedHashMap<String, CustomFieldV2> target) {
    CustomFieldV2 bf = CustomFieldV2.buildField(oriBf.getFieldName(), oriBf.getValue(), oriBf.getSqlType(), oriBf.isKey(), oriBf.isNull(), oriBf.isUpdated());
    target.put(oriBf.getFieldName(), bf);
}

private static void processOrderExtend1InsertAndUpdateToUpdate(CustomData data, List<CustomData> re, EventTypeInSdk eventType) {
    for (CustomRecordV2 oriRecord : data.getRecords()) {
        CustomRecordV2 updateRecord = new CustomRecordV2();

        if (eventType.equals(EventTypeInSdk.INSERT)) {
            // 需要处理 afterColumnMap > beforeColumnMap
            for (Map.Entry<String, CustomFieldV2> f : oriRecord.getAfterColumnMap().entrySet()) {
                changeOrderExtend1FillColumnMap(f.getValue(), updateRecord.getBeforeColumnMap());
            }

            for (Map.Entry<String, CustomFieldV2> f : oriRecord.getAfterKeyColumnMap().entrySet()) {
                changeOrderExtend1FillColumnMap(f.getValue(), updateRecord.getBeforeKeyColumnMap());
            }
        }

        if (eventType.equals(EventTypeInSdk.UPDATE)) {
            // 需要处理 beforeColumnMap
            for (Map.Entry<String, CustomFieldV2> f : oriRecord.getBeforeColumnMap().entrySet()) {
                changeOrderExtend1FillColumnMap(f.getValue(), updateRecord.getBeforeColumnMap());
            }

            for (Map.Entry<String, CustomFieldV2> f : oriRecord.getBeforeKeyColumnMap().entrySet()) {
                changeOrderExtend1FillColumnMap(f.getValue(), updateRecord.getBeforeKeyColumnMap());
            }
        }

        for (Map.Entry<String, CustomFieldV2> f : oriRecord.getAfterColumnMap().entrySet()) {
            changeOrderExtend1FillColumnMap(f.getValue(), updateRecord.getAfterColumnMap());
        }

        for (Map.Entry<String, CustomFieldV2> f : oriRecord.getAfterKeyColumnMap().entrySet()) {
            changeOrderExtend1FillColumnMap(f.getValue(), updateRecord.getAfterKeyColumnMap());
        }

        List<CustomRecordV2> updateRecords = new ArrayList<>();
        updateRecords.add(updateRecord);

        SchemaInfo updateSchemaInfo = data.cloneSchemaInfo(data.getSchemaInfo());
        CustomData updateData = new CustomData(updateSchemaInfo, EventTypeInSdk.UPDATE, updateRecords);

        re.add(updateData);
    }
}

工程项目代码参考：

cloudcanal-data-process 本工程汇集了 CloudCanal 数据处理插件，以达成数据自定义 transformation 目标

https://gitee.com/clougence/cloudcanal-data-process

问题！
insert改写为update在增量同步阶段工作正常，在全量迁移阶段会报错！
截止2023年3月代码中没有办法判断任务阶段是在全量迁移还是增量同步，所以只能在第一次建立任务的时候不要改写insert成为update，待任务正常执行时候，再重新上传激活代码处理包，
让insert改写update在增量同步阶段工作
由于全量迁移阶段的insert数据没同步，这个暂时只有单独手动处理更新到目标表去

2023-03-14 14:22:50.791 [full-task-executor-7-thd-1] INFO  c.c.c.b.service.task.parser.full.RdbSinglePkPageScanner - [FINISH SCAN!]null.你的库.old_orders_extend1, cost time 222 ms.migrated count is 1774
2023-03-14 14:22:51.660 [full-apply-disruptor-6-thd-3] ERROR c.c.c.task.applier.full.FullDisruptorExceptionHandler - disruptor process full event error,msg:IllegalArgumentException: unsupported one FullEvent with multi CustomData.
java.lang.IllegalArgumentException: unsupported one FullEvent with multi CustomData.
 at com.clougence.cloudcanal.task.data.process.pkg.editorv2.FullDataDeSerializer.deserialize(FullDataDeSerializer.java:59)
 at com.clougence.cloudcanal.task.data.process.pkg.editorv2.FullDataDeSerializer.deserialize(FullDataDeSerializer.java:18)
 at com.clougence.cloudcanal.task.data.process.pkg.V2PkgProcessor.process(V2PkgProcessor.java:125)
 at com.clougence.cloudcanal.task.data.process.pkg.CustomPkgProcessor.process(CustomPkgProcessor.java:73)
 at com.clougence.cloudcanal.task.data.process.DataProcHandlerImpl.process(DataProcHandlerImpl.java:19)
 at com.clougence.cloudcanal.task.data.process.FullDataProcWorkerHandler.onEvent(FullDataProcWorkerHandler.java:32)
 at com.clougence.cloudcanal.task.data.process.FullDataProcWorkerHandler.onEvent(FullDataProcWorkerHandler.java:12)
 at com.lmax.disruptor.WorkProcessor.run(WorkProcessor.java:143)
 at java.lang.Thread.run(Thread.java:748)

3.3 CloudCanal任务建立

事实主表建立一个任务实例

old_orders 到 new_orders，并配置好字段映射

业务延迟产生数据的表建立一个任务实例

old_orders_extend1 到 new_orders，并配置好字段映射

old_orders_extend2 到 new_orders，并配置好字段映射

建立任务时在配置时候数据处理界面上传自定义代码的jar包

自定义代码jar包更新

激活后需要重启任务（PS：需要重启至少2次才生效，这个是CloudCanal bug，截止2023-3-10 还未解决）

3.4 实现结果

源表的insert和update都会准实时的同步到目标表达到设计目标

流水理鱼

流水理鱼（wwek）的博客

标签： cdc

CloudCanal在表重构中的应用MySQL三表合一准实时同步