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

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都会准实时的同步到目标表达到设计目标

发表回复