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

2024-01-03 12:01:38

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工作过程:

53ae5ece3e541795e375231e20028ad8.jpeg

消息生产者并没有直接将消息发送给消息队列,而是通过建立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_%';

04d70d4eaa725f5fda21994b8024b511.jpeg

(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用户如下:

094ccb6681e3dde6397fc0b9b7612129.jpeg

9cb795035d943d6585fccbd37f79e568.jpeg

(2)新建交换机

1d1eb5345911390cf92dbf4b3b83131e.jpeg

(3)新建队列

d5403ca77736c4f1db2ccffb3026f4cb.jpeg

(4)绑定交换机和队列

点击下图红框处

7f5eab365ac579c4d19cdb266c564fca.jpeg

绑定交换机和队列

cf1f0cee32719b807737990436b70592.jpeg

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

0f02ced09254970f7d9df40b0c5f0643.jpeg

(2)配置

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

配置canal.properties

选择模式

49af3498f72fa9910763cddf0971d47a.jpeg

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

0b1f68ce56ab3bb7bfd5ad807a5d884c.jpeg

设置RabbitMQ相关属性

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

rabbitmq.deliveryMode = 2(2表示Durable持久化)

31bb00e87644bc397a2b5a0f5100e6d2.jpeg

配置instance.properties

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

af5f7868ceeda58c37d1347f1d676d13.jpeg

设置用户和密码

504ba1017b1a4a1e57f426d3d178565e.jpeg

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

63797eaa89de3ee92722e61ade952e04.jpeg

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

4e68543d75543e4cad58756f627eeb42.jpeg

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

619e1973d92dccc63a26821c4849de94.jpeg

在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

选择模式

bdb2676c03b091905fba830f6a9e2013.jpeg

rabbitMQ消费者配置

dece93d7feade3521957d5d7514e5df6.jpeg

源数据库配置

2c5f872b9e7caeffbddb13c95f640f64.jpeg

目标数据库配置

instance: ck-queue配置rabbitmq的队列

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

97839059523f792a424aa17a54cb7c21.jpeg

配置mytest_user.yml

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

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

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

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

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

b8b7eef64226b38e929481a7a02b4ebf.jpeg

结果演示

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

启动canal,在canal目录下:

bin/startup.sh

查看server日志:

tail -200f logs/canal/canal.log

e88344db81fee160471724fe61bb3f14.jpeg

查看instance日志:

tail -200f logs/example/example.log

a1e3c7996eb512786d4858562cd1175f.jpeg

启动canal-client,在canal-client目录下:

bin/startup.sh

canal-client目录下查看日志:

tail -200f logs/adapter/adapter.log

af87e706261538fb87d6a294b3d41f75.jpeg

向MySQL的test表中写入数据

source /opt/test.sql;

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

03e54ff9bed72df253f0c84437594a59.jpeg

进入ClickHouse中查看数据

75d1abe47d7fceb08cd9cbcb9d5843a1.jpeg

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

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

进入ClickHouse中查看数据

0b6639f4db2f39b0dff9ea43595eb06e.jpeg

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

获取更多内容,欢迎关注万山数据!

文章来源:https://blog.csdn.net/zjyy1987/article/details/135357870
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。