Ali Canal实现Mysql数据采集转储

近期接触的比较多的关于mysql数据实时采集转储，以及后续计算分析的项目，这里对使用到的canal框架进行介绍、基于canal框架的架构设计以及升级canal1.1.0之后的架构优化。

ali canal介绍

canal 是阿里巴巴mysql数据库binlog的增量订阅&消费组件。早起是为了解决ali跨机房同步的需求。canal分内部版本以及外部版本，外部版本即开源版本，仅支持mysql以及mysql核心（mariadb）5.7及以下的版本.

基于日志增量订阅&消费支持的业务：

• 数据库镜像

• 数据库实时备份

• 多级索引 (卖家和买家各自分库索引)

• search build

• 业务cache刷新

• 价格变化等重要业务消息

工作原理

mysql主备复制实现

从上层来看，复制分成三步：

master将改变记录到二进制日志(binary log)中（这些记录叫做二进制日志事件，binary log events，可以通过show binlog events进行查看）；
slave将master的binary log events拷贝到它的中继日志(relay log)；
slave重做中继日志中的事件，将改变反映它自己的数据。

canal的工作原理

原理相对比较简单：

• canal模拟mysql slave的交互协议，伪装自己为mysql slave，向mysql master发送dump协议

• mysql master收到dump请求，开始推送binary log给slave(也就是canal)

• canal解析binary log对象(原始为byte流)

架构设计

需求

其实需求很简单，我们需要将数据从mysql抽取，将数据存储到kafka中，提供给后续程序的存储或者分析计算，但是时效性上要求实时。

需求分析

根据上述的需求理解，在做整体设计的时候，从kafka获取数据进行计算存储这块已经比较成熟了，所以我们更多的是要解决前面的从mysql采集数据存放到kafka的问题。
从mysql获取数据，整体来说有两种方法：

• 第一种，可以直接从mysql读取新数据，不过这个方法一方面你要去判断新数据是否为新数据，另一方面还要考虑这种读取方法是否会对其他用户的访问造成影响，还有就是要做到实时很难。

• 第二种，通过解析mysql binlog方式来解决实时数据抽取存储的问题，canal则可以满足这一点，不过canal无法直接存储到kafka中，需要通过其他手段将数据存入到kafka中

架构图

上图中，mysql数据通过binlog机制，生成binlog文件，通过canal进行binlog的解析，将数据通过canal source传入到flume，再通过flume写入到kafka，然后再提供给后续程序的存储以及分析计算。这里就不画的那么详细了，我们的重点放在前面的数据采集。其中canal source是需要自己自定义开发的canal source插件。

我们的标题中提到了要对上述架构的优化，其实这个优化主要由于canal新版本发布提供的一些很给力的功能，让我们的整体框架上得到了较好的改进，接下来我们先看下canal新版的介绍。

canal 1.1.0介绍

重要功能优化

• 整体性能测试&优化,提升了150%.

• 表结构TSDB相关问题全部修复(比较多的是DDL语法解析兼容性)

• 基于bio修复binlog长时间无数据导致的半开链接无法恢复

功能新增

• 原生支持prometheus监控

• 原生支持kafka消息投递

• 原生支持aliyun rds的binlog订阅 (解决自动主备切换/oss binlog离线解析) 参考: 【Aliyun RDS QuickStart】

• 原生支持docker镜像

• MySQL gtid模式订阅支持

• MySQL XA binlog事件

• Mysql Show Slave Hosts状态支持

基于canal 1.1.0的架构优化

其实这里我们最关注的功能其实是对于kafka的支持，因为在我们前面的架构设计过程中，由于数据从canal抽取之后无法直接存入kafka，所以需要我们自己读取数据写入到kafka中，有了这个功能我们可以大大简化我们的前期数据采集存储流程。

优化后架构图

采用canal1.1.0之后，整体架构就变成上图的样子，优化了从canal-flume-kafka的中间环节，直接通过canal写入到kafka topic中。整体优化之后我们既不需要自定义开发canal flume source，也不需要考虑使用flume过程中遇到的问题。