1 概述
MySQL 的用户群体很大，为了能够增强数据的实时性，很多解决方案会利用 binlog 将数据写入到 ClickHouse。为了能够监听 binlog 事件，我们需要用到类似 canal 这样的第三方中间件，这无疑增加了系统的复杂度。
ClickHouse 20.8.2.3版本新增加了 MaterializeMySQL 的 database 引擎，该 database 能映射到 MySQL 中的某个 database，并自动在 ClickHouse 中创建对应的 ReplacingMergeTree。ClickHouse服务做为MySQL副本，读取Binlog并执行DDL和DML请求，实现了基于MySQL Binlog机制的业务数据库实时同步功能。
1.1 特点
（1）MaterializeMySQL 同时支持全量和增量同步，在 database 创建之初会全量同步 MySQL 中的表和数据，之后则会通过 binlog 进行增量同步。
（2）MaterializeMySQL database 为其所创建的每张 ReplacingMergeTree 自动增加了 _sign 和 _version 字段。
其中， _version 用作 ReplacingMergeTree 的 ver 版本参数，每当监听到 insert、update 和 delete 事件时，在 databse 内全局自增。而 _sign 则用于标记是否被删除，取值 1 或者 -1。
目前 MaterializeMySQL 支持如下几种 binlog 事件:
- MYSQL_WRITE_ROWS_EVENT: _sign = 1，_version ++
- MYSQL_DELETE_ROWS_EVENT: _sign = -1，_version ++
- MYSQL_UPDATE_ROWS_EVENT: 新数据 _sign = 1
- MYSQL_QUERY_EVENT: 支持 CREATE TABLE 、DROP TABLE 、RENAME TABLE等。
1.2 使用细则
（1）DDL查询
MySQL DDL查询被转换成相应的ClickHouse DDL查询（ALTER, CREATE, DROP, RENAME）。如果ClickHouse不能解析某些DDL查询，该查询将被忽略。
（2）数据复制
MaterializeMySQL不支持直接插入、删除和更新查询，而是将DDL语句进行相应转换：
MySQL INSERT查询被转换为INSERT with _sign=1。
MySQL DELETE查询被转换为INSERT with _sign=-1。
MySQL UPDATE查询被转换成INSERT with _sign=1和INSERT with _sign=-1。
（3）SELECT查询
如果在SELECT查询中没有指定_version，则使用FINAL修饰符，返回_version的最大值对应的数据，即最新版本的数据。
如果在SELECT查询中没有指定_sign，则默认使用WHERE _sign=1，即返回未删除状态（_sign=1)的数据。
（4）索引转换
ClickHouse数据库表会自动将MySQL主键和索引子句转换为ORDER BY元组。
ClickHouse只有一个物理顺序，由ORDER BY子句决定。如果需要创建新的物理顺序，请使用物化视图。
2 案例实操
2.1 MySQL开启binlog和GTID模式
（1）确保 MySQL 开启了 binlog 功能，且格式为 ROW
打开/etc/my.cnf,在[mysqld]下添加：
server-id=1
log-bin=mysql-bin
binlog_format=ROW
（2）开启GTID模式
如果如果clickhouse使用的是20.8 prestable之后发布的版本，那么MySQL还需要配置开启GTID模式, 这种方式在mysql主从模式下可以确保数据同步的一致性(主从切换时)。
gtid-mode=on
enforce-gtid-consistency=1 # 设置为主从强一致性
log-slave-updates=1 # 记录日志
GTID 是 MySQL 复制增强版，从 MySQL 5.6 版本开始支持，目前已经是 MySQL 主流复制模式。它为每个 event 分配一个全局唯一ID和序号，我们可以不用关心 MySQL 集群主从拓扑结构，直接告知 MySQL 这个 GTID 即可。
（3）重启MySQL
sudo systemctl restart mysqld
2.2 准备MySQL表和数据
（1）在 MySQL 中创建数据表并写入数据
CREATE DATABASE testck;
CREATE TABLE `testck`.`t_organization` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`code` int NOT NULL,
`name` text DEFAULT NULL,
`updatetime` datetime DEFAULT NULL,
PRIMARY KEY (`id`),
UNIQUE KEY (`code`)
) ENGINE=InnoDB;
INSERT INTO testck.t_organization (code, name,updatetime) VALUES(1000,'Realinsight',NOW());
INSERT INTO testck.t_organization (code, name,updatetime)
VALUES(1001, 'Realindex',NOW());
INSERT INTO testck.t_organization (code, name,updatetime) VALUES(1002,'EDT',NOW());
（2）创建第二张表
CREATE TABLE `testck`.`t_user` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`code` int,
PRIMARY KEY (`id`)
) ENGINE=InnoDB;
INSERT INTO testck.t_user (code) VALUES(1);
2.3 开启ClickHouse物化引擎
set allow_experimental_database_materialize_mysql=1;
2.4 创建复制管道
（1）ClickHouse中创建 MaterializeMySQL 数据库
CREATE DATABASE test_binlog ENGINE = MaterializeMySQL('hadoop1:3306','testck','root','000000');
其中 4 个参数分别是 MySQL地址、databse、username 和 password。
（2）查看ClickHouse的数据
use test_binlog;
show tables;
select * from t_organization;
select * from t_user;
2.5 修改数据
（1）在 MySQL 中修改数据:
update t_organization set name = CONCAT(name,'-v1') where id = 1
（2）查看clickhouse日志可以看到binlog监听事件，查询clickhouse
select * from t_organization;
2.6 删除数据
（1）MySQL删除数据:
DELETE FROM t_organization where id = 2;
（2）ClicKHouse，日志有 DeleteRows 的 binlog 监听事件，查看数据：
select * from t_organization;
（3）在刚才的查询中增加 _sign 和 _version 虚拟字段
select *,_sign,_version from t_organization order by _sign desc,_version desc;
在查询时，对于已经被删除的数据，_sign=-1，ClickHouse 会自动重写 SQL，将 _sign = -1 的数据过滤掉;
对于修改的数据，则自动重写 SQL，为其增加 FINAL 修饰符。
select * from t_organization
等同于
select * from t_organization final where _sign = 1
2.7 删除表
（1）在mysql执行删除表
drop table t_user;
（2）此时在clickhouse处会同步删除对应表，如果查询会报错
show tables;
select * from t_user;
DB::Exception: Table scene_mms.scene doesn't exist..
（3）mysql新建表，clickhouse可以查询到
CREATE TABLE `testck`.`t_user` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`code` int,
PRIMARY KEY (`id`)
) ENGINE=InnoDB;
INSERT INTO testck.t_user (code) VALUES(1);
#ClickHouse查询
show tables;
select * from t_user;
