实战 | MySQL Binlog通过Canal同步HDFS

王知无-import_bigdata

发布于 2020-02-20 10:07:54

2.5K0

发布于 2020-02-20 10:07:54

文章被收录于专栏：大数据成神之路大数据成神之路大数据成神之路

之前《MySQL Binlog同步HDFS的方

本文来自：http://bigdatadecode.club/MysqlToHDFSWithCanal.html

canal server 部署

在canal中一个mysql实例对应一个配置文件，配置文件放在conf目录下的一个文件夹中，该文件夹的名字就代表了mysql实例。结构如下

-rwxr-xr-x 1 dc user 2645 Jul 18 14:25 canal.properties
-rwxr-xr-x 1 dc user 2521 Jul 17 18:31 canal.properties.bak
-rwxr-xr-x 1 dc user 3045 Jul 17 18:31 logback.xml
drwxr-xr-x 2 dc user 4096 Jul 17 18:38 spring
drwxr-xr-x 2 dc user 4096 Jul 19 11:55 trans1

trans1代表一个mysql实例，该文件夹中有个instance.properties文件，在里面配置mysql数据库的信息。

## mysql serverId 部署ha的话，slaveId不能重复
canal.instance.mysql.slaveId = 1235
canal.instance.master.address = 10.172.152.66:3306
# username/password
canal.instance.dbUsername = root
canal.instance.dbPassword = root
# 采集表的正则
canal.instance.filter.regex = .*\\..*

canal server HA部署

采用canal的HA模式，canal的HA是依赖zk来实现的。修改配置文件，配置文件是conf目录下大canal.properties

# common argument
canal.id= 1             # 另一台 canal.id= 2
canal.ip= 10.172.152.66 # 另一台 canal.ip= 10.172.152.124
canal.port= 11111
canal.zkServers= 10.172.152.66:2181,10.172.152.124:2181,10.172.152.125:2181
# conf目录下mysql实例的文件名
canal.destinations= trans1
# 使用ha必须使用default-instance.xml
canal.instance.global.spring.xml = classpath:spring/default-instance.xml

canal的部署比较简单，在instance.properties也设置从哪个binlog某个位置处开始读(未测)。下面看下client的设计，由于canal并没有继承client只是提供了一套client的api由用户自己去实现，则这里重点记录下client的设计。

canal client 功能设计

client的主要功能是与canal server的某个destinations建立连接消费订阅的binlog信息，并将binlog进行解析落地到存储系统中。

client消费原理

client api中有ack和rollback机制，保证了数据不丢失。 ack机制采用异步确认，也就是可以连续调用get多次，后续异步按顺序提交ack/rollback，这种机制在canal中称为流式api设计。

流式api设计的好处：

get/ack异步化，减少因ack带来的网络延迟和操作成本 (99%的状态都是处于正常状态，异常的rollback属于个别情况，没必要为个别的case牺牲整个性能)
get获取数据后，业务消费存在瓶颈或者需要多进程/多线程消费时，可以不停的轮询get数据，不停的往后发送任务，提高并行化。

数据格式及使用

client将binlog中的信息解析成json，json格式分为两种，一种是DML，主要包括insert、update和delete，另一种是DDL，主要包括create、alter和drop。 json格式如下：

// DML-json
{
   "database":"test",
   "table":"e",
   "type":"delete/insert/update",
   "executeTime":"1501645930000",
   "consumerTime":"1501645930000",
   "data":[
   		{"id":"1","num":"1"},
   		{"id":"2","num":"2"}
   ]
}
// DDL-json
{
   "database":"test",
   "table":"e",
   "type":"create/alter/drop",
   "executeTime":"1501645930000",
   "consumerTime":"1501645930000",
   "isDDL":"true/false",
   "sql":"create table e(id int)"
}
之所以分为两个json格式，是因为想把DML-json里的数据直接格式化作为当天的增量使用，而DDL-json则需要解析成hql，个人建议在执行时需要人工check，DDL-json每天应该不会太多，初期人工check的压力不大，等程序成熟之后可以不用人工check。

解析DML-json的时候需要给数据新增一列来标识数据的状态，数据的状态是指数据是否被删除。

配置

方便程序的移植性将一些参数提出，作为配置文件，由程序动态周期性的加载。

ZK_HOSTS = # canal server ha的zk地址 DESTINATION = # 要消费的mysql实例 FILTER_REGEX = # 订阅的表信息的正则 BATCH_SIZE = # 批量获取的条数 DML_PATH = DDL_PATH = DELAY_TIME = # 延迟报警的阈值，单位ms

多线程异步处理数据

client api提供get／ack／rollback接口，为了提高消费效率，采用异步ack的机制。

设计一个一定大小的ackQueue，get不断的获取数据，将message交给新线程处理并将batchId放入ackQueue中，待新线程处理完message之后进行ack确认，从ackQueue中取出batchId按顺序确认，如遇到异常进行回滚。

伪代码如下：

try{
	while(true){
		Message message = connector.getWithoutAck(BATCH_SIZE); // 获取指定数量的数据
        batchId = message.getId();
        ackQueue.put(batchId);
        executorService.submit(new Runnable() {
            public void run() {
            	// 需要一个线程和文件的映射，防止多个线程写同一个文件
                parseMessage(message);
                // 判断batchId是否和ackQueue中取得的batchId一致，大于则等待小于报异常
                ackMessage(batchId);
            }
        });
	}
}catch (Exception e){
	connector.rollback(batchId);
}finally {
    ...
}