Canal+RabbitMQ实现MySQL数据同步至ClickHouse

ClickHouse作为一个被广泛使用OLAP分析引擎，在执行分析查询时的速度优势很好的弥补了MySQL的不足，但是如何将MySQL数据同步到ClickHouse就成了用户面临的第一个问题。本文利用Canal来实现ClickHouse实时同步MySQL数据，使用RabbitMQ来做消息队列，给出了将MySQL多张表同步至ClickHouse同一张表的方案。

Canal简介;

Canal主要用途是基于MySQL数据库增量日志解析，提供增量数据订阅和消费。

工作原理：

·Canal模拟MySQL slave的交互协议，伪装自己为MySQL slave，向MySQL master发送dump协议

·MySQL master收到dump请求，开始推送binary log给slave(即Canal)

·Canal解析binary log对象(原始为byte流)

RabbitMQ简介

RabbitMQ是一款使用Erlang语言开发的，实现AMQP(高级消息队列协议)的开源消息中间件。

RabbitMQ工作过程：

消息生产者并没有直接将消息发送给消息队列，而是通过建立Exchange（交换器）和Channel（信道），将消息发送给Exchange，Exchange根据routing key，将消息转发给指定的Queue(消息队列)。然后，消息会被消费者从队列里读取并消费。

接下来让我们开始进入实操

MySQL及ClickHouse建表示例

MySQL示例：

CREATE TABLE `test` (

`id` int(11) NOT NULL,

?`name` varchar(100) DEFAULT NULL,

?`quantity` int(11) DEFAULT NULL,

PRIMARY KEY (`id`)

)

ClickHouse示例：

CREATE TABLE default.test

(

?`id` Int32,

`name` String,

`quantity` Int32

)

ENGINE = MergeTree

PRIMARY KEY id

ORDER BY id

MySQL配置

（1）开启binlog

vi /etc/my.cnf 添加下面内容：

server-id = 1

log_bin = /var/lib/mysql/bin.log

binlog-format = row

expire_logs_days = 30

max_binlog_size= 768M

bind-address = 0.0.0.0

重启MySQL服务：

systemctl restart mysqld.service

登陆mysql，查看binlog启动情况：

show variables like 'log_%';

（2）新增同步账号

登陆mysql,执行下面命令，创建账号maxwell，密码为123456

CREATE USER 'maxwell'@'%' IDENTIFIED BY '123456';GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO 'maxwell'@'%';flush privileges;

rabbitMQ配置

登录http://IP:15672，进入rabbitMQ管理页面（账号和密码默认都是guest）

（1）新建同步用户

注1：不要使用初始guest账户，在canal连接时，权限会不够。

注2：密码中除了下划线，不要使用特殊字符，否则canal消费端配置会报错。

新建cktest用户如下：

（2）新建交换机

（3）新建队列

（4）绑定交换机和队列

点击下图红框处

绑定交换机和队列

canal服务端安装配置

（1）安装

下载canal安装包

wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.deployer-1.1.5.tar.gz

/opt下创建canal目录

mkdir canal

解压到指定目录

tar zxvf canal.deployer-1.1.5.tar.gz -C canal

（2）配置

服务端要配置两个文件conf文件下canal.properties、example文件夹中的instance.properties

配置canal.properties

选择模式

配置读取mysql二进制文件的用户名和密码

设置RabbitMQ相关属性

rabbitmq.exchange 填写mq队列相对应的交换机名称

rabbitmq.deliveryMode = 2（2表示Durable持久化）

配置instance.properties

配置MySQL数据库的IP地址和端口

设置用户和密码

表过滤 (.*\\..*)所有库所有表  (xxx\\..*)指定库所有表 (xxx\\.xxx)指定库指定表，如果多个用英文逗号隔开

canal.mq.topic=example-routingkey 配置交换机和队列的routingkey

canal客户端安装配置

（1）安装

下载canal安装包

wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.adapter-1.1.5.tar.gz

/opt下创建canal-client目录

mkdir canal-client

解压到指定目录

tar zxvf canal.adapter-1.1.5.tar.gz -C canal-client

在lib目录下要导入几个包：

clickhouse-jdbc-qbe-0.2.4-jar-with-dependencies.jar

httpclient-4.5.5.jar

httpcore-4.4.9.jar

lz4-1.3.0.jar

lz4-java-1.4.1.jar

（2）配置

canal客户端配置包含两部分。application.yml（应用配置）和rdb文件夹中xxx.yml配置（数据映射配置）

配置application.yml

选择模式

rabbitMQ消费者配置

源数据库配置

目标数据库配置

instance: ck-queue配置rabbitmq的队列

key：example-routingkey填写mq队列的key

配置mytest_user.yml

注：一个yml文件，仅可编辑一张表的映射关系，多张表就要新建多个配置文件。

dataSourceKey: 对应application.yml中的配置，默认为defaultDS

destination: 对应application.yml中的instance配置

targetTable: 目标库的目标表，不需要带数据库名称，否则会出现：库名.库名.表名的错误

mapAll: 映射关系true为全映射，false为非全映射，若为非全映射，则需要编辑targetColumns下面的配置来进行字段映射

结果演示

我们配置了mytest_user.yml、test1.yml两个文件，将MySQL中的两个表同步至ClickHouse。

启动canal，在canal目录下：

bin/startup.sh

查看server日志：

tail -200f logs/canal/canal.log

查看instance日志：

tail -200f logs/example/example.log

启动canal-client，在canal-client目录下：

bin/startup.sh

canal-client目录下查看日志：

tail -200f logs/adapter/adapter.log

向MySQL的test表中写入数据

source /opt/test.sql;

可以看到canal客户端会输出如下日志：

进入ClickHouse中查看数据

再向MySQL的test1表中写入数据：

insert into test1 values(10001,'apple',13);

进入ClickHouse中查看数据

可以看到数据已经同步至ClickHouse。

获取更多内容，欢迎关注万山数据！