前言

已经有将近仨月没更新博客了, 这三个月我换了两份工作了. 最近入职的这家公司让我搞数据异构同步的方案和逻辑. 因为数据清洗是相当复杂的, 再加上要熟悉公司的业务和技术, 所以这三个月一直没有来得及更新.
废话不多说了, 直接来看看我们的需求.

需求描述

我们需要收集数据进行分析, 因此需要从其他的系统中获取数据. 根据数据种类, 一共分为了3种操作:

使用数据同步工具, 直接监听mysql的binlog进行数据同步.
提供接口, 让上游系统定时调用我们的接口, 传递数据.
定时主动调用上游系统的接口进行数据的获取.

这三种方式里, 2和3都没有什么说明的意义, 毕竟只要学完了java就能做.
那么这里主要说明第一种方式, binlog增量同步.

设计

由于公司倾向于使用云服务, 因此首先考虑的是各家厂商的云.
最后由于报价问题, 选择了某某云. (就是谁便宜用谁啦)
也就是说, 目前有两种方案:

某某云的数据同步服务和数据订阅服务. (主要)
自己搭建canal服务器和kafka集群. (备用)

任务说明 (肯定不能太详细)

由于我们现有系统使用了别的全量同步, 而且用到了rabbitmq. 因此现在首先考虑的问题是, 基于云服务的环境下, rabbitmq是否可以被替代.
首先, 我考虑了最近大火的pulsar.
于是开始从0学习pulsar, 发现pulsar真的是一个好东西. 从2.3.0版本开始支持canal, 不仅可以作为队列使用, 还可以完成数据同步的任务需求, 性能稳定, 吞吐量极高. 甚至, 我已经封装好一部分代码了, 可以投入使用了.
但是!!!
由于云服务并不支持, 所以放弃pulsar.

-----------------------------------华丽的分割线-------------------------------

接下来, 就是任务的重点了, 使用云服务自带的kafka进行数据同步.
然后又是一顿看文档…
看完文档, 就是花钱买服务, 然后看demo写代码测试.

值得一提的是, 这个服务的kafka使用的是谷歌的protobuf协议, 但是和谷歌推荐的做法完全不同.
谷歌封装了一个工具, 用来生成对应实体的协议类, 通过这个协议类将从kafka接收到的数据转换成对应实体.
而这家云服务商, 直接进行了封装, 让所有的数据都变成一个对象, 最后我们拿到的就是这个event对象, 只需要解析这个event对象即可拿到数据. (这里还有不止一个坑)

调研数据同步的踩坑记录

首先就是调试问题.
该项服务不支持外网连接, 同时使用的是docker容器, 因此既无法本地调试, 也无法远程调试. 只能不断的重新集成代码和大量的log测试.
数据封装问题.
谷歌提供的做法虽然前期比较麻烦, 但是好处在于不需要手写代码转换为各种对象.但是该服务封装的代码中, 数据被分为了3部分. columns, newDataColumns, oldDataColumns. 导致列名和对应的值无法匹配, 需要至少2层循环才能获取到数据. (仔细考虑过之后, 发现这么封装了也没啥问题. 毕竟来源数据结构未知. 不过若是封装成map也是没有问题的. 其实也可以不用他的工具类, 我是这么觉得.)
源表的结构问题
这一点可就有意思了. 上游要对接各部门的系统, 因此表的结构和命名也千奇百怪. 有的列名纯大写, 有的列名纯小写. 有的表名不带下划线. 有的表有主键, 有的表啥约束都没有(?).
这个地方遇到的坑就和约束有关系.
当表中含有主键时, 做一个update操作, 即使条件不适用主键, 最终也会转换为主键写入binlog (这点有待证实, 因为我只拿到了结果是主键id, 而拿不到其他数据, 因此推测是这样).
当表中不含主键是, 做一个update操作, 条件是什么数据, 那么拿到的就是什么数据.
使用设计模式
其实这个说是问题吧也不是问题, 说不是问题吧, 也费了点功夫去写.
由于数据封装的有问题, 因此在程序入口处拿到kafka的消息后, 使用策略模式传入数据context对象, 调用具体的策略类进行数据清洗.