clickhouse通过MaterializeMySQL库引擎实现mysql实时同步

￡神魔★判官ぃ 2024-04-01 18:51 108阅读 0赞

#### 文章目录 ####

*  一.概述
 *   *  1.特点
     *  2.使用细则
     *   *  （1）DDL 查询
         *  （2）数据复制
         *  （3）SELECT 查询
     *  （4）索引转换
     *   *  （5）类型转换
         *  （6）创建语句以及配置参数
         *  （7）表覆盖
         *  （8）注意事项
 *  二.案例实操
 *   *  1. MySQL 开启 binlog 和 GTID 模式
     *   *  （1）确保 MySQL 开启了 binlog 功能，且格式为 ROW
         *  （2）开启 GTID 模式
         *  （3）重启 MySQL
         *  （4）用户
     *  2. 准备 MySQL 表和数据
     *   *  （1）在 MySQL 中创建数据表并写入数据
         *  （2）创建第二张表
     *  3. 开启 ClickHouse 物化引擎
     *  4. 创建复制管道
     *   *  （1）ClickHouse 中创建 MaterializeMySQL 数据库
         *  （2）查看 ClickHouse 的数据
     *  5. 修改数据
     *   *  （1）在 MySQL 中修改数据:
         *  （2）查看 clickhouse 日志可以看到 binlog 监听事件，查询 clickhouse
     *  6. 删除数据
     *   *  （1）MySQL 删除数据:
         *  （2）ClicKHouse，日志有 DeleteRows 的 binlog 监听事件，查看数据：
         *  （3）在刚才的查询中增加 \_sign 和 \_version 虚拟字段
     *  7. 删除表
     *   *  （1）在 mysql 执行删除表
         *  （2）此时在 clickhouse 处会同步删除对应表，如果查询会报错
         *  （3）mysql 新建表，clickhouse 可以查询到

## 一.概述 ##

为了能够增强数据的实时性，利用 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）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 等。

即支持mysql 5.6/5.7/8.0版本数据库，兼容insert，update，delete，alter，create，drop，truncate等大部分DDL操作。

### 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。

即使用MaterializedMySQL数据库引擎时，ReplacingMergeTree表与虚拟\_sign和\_version列一起使用。

*  \_version— 交易计数器。键入UInt64。
 *  \_sign— 删除标记。键入Int8。可能的值:
    
     *  1— 未删除行，
     *  \-1— 行被删除。

#### （3）SELECT 查询 ####

如果在 SELECT 查询中没有指定\_version，则使用 FINAL 修饰符，返回\_version 的最大值对应的数据，即最新版本的数据。

如果在 SELECT 查询中没有指定\_sign，则默认使用 WHERE \_sign=1，即返回未删除状态（\_sign=1)的数据。

### （4）索引转换 ###

ClickHouse 数据库表会自动将 MySQL 主键和索引子句转换为 ORDER BY 元组。

ClickHouse 只有一个物理顺序，由 ORDER BY 子句决定。如果需要创建新的物理顺序，请使用物化视图。

*  `_sign=-1`没有从表中物理删除的行。
 *  `UPDATE/DELETE`引擎不支持级联查询`MaterializedMySQL`，因为它们在 MySQL 二进制日志中不可见。
 *  复制很容易被破坏。
 *  禁止对数据库和表进行手动操作。
 *  `MaterializedMySQL`受optimize\_on\_insert 设置影响。`MaterializedMySQL`当 MySQL 服务器中的表发生变化时，数据会合并到数据库中的相应表中。

#### （5）类型转换 ####

<table> 
 <thead> 
  <tr> 
   <th align="center">MySQL</th> 
   <th align="right">ClickHouse</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="center">TINY</td> 
   <td align="right">Int8</td> 
  </tr> 
  <tr> 
   <td align="center">SHORT</td> 
   <td align="right">Int16</td> 
  </tr> 
  <tr> 
   <td align="center">INT24</td> 
   <td align="right">Int32</td> 
  </tr> 
  <tr> 
   <td align="center">LONG</td> 
   <td align="right">UInt32</td> 
  </tr> 
  <tr> 
   <td align="center">LONGLONG</td> 
   <td align="right">UInt64</td> 
  </tr> 
  <tr> 
   <td align="center">FLOAT</td> 
   <td align="right">Float32</td> 
  </tr> 
  <tr> 
   <td align="center">DOUBLE</td> 
   <td align="right">Float64</td> 
  </tr> 
  <tr> 
   <td align="center">DECIMAL, NEWDECIMAL</td> 
   <td align="right">Decimal</td> 
  </tr> 
  <tr> 
   <td align="center">DATE, NEWDATE</td> 
   <td align="right">Date32</td> 
  </tr> 
  <tr> 
   <td align="center">DATETIME, TIMESTAMP</td> 
   <td align="right">DateTime</td> 
  </tr> 
  <tr> 
   <td align="center">DATETIME2, TIMESTAMP2</td> 
   <td align="right">DateTime64</td> 
  </tr> 
  <tr> 
   <td align="center">YEAR</td> 
   <td align="right">UInt16</td> 
  </tr> 
  <tr> 
   <td align="center">TIME</td> 
   <td align="right">Int64</td> 
  </tr> 
  <tr> 
   <td align="center">ENUM</td> 
   <td align="right">Enum</td> 
  </tr> 
  <tr> 
   <td align="center">STRING</td> 
   <td align="right">String</td> 
  </tr> 
  <tr> 
   <td align="center">VARCHAR, VAR_STRING</td> 
   <td align="right">String</td> 
  </tr> 
  <tr> 
   <td align="center">BLOB</td> 
   <td align="right">String</td> 
  </tr> 
  <tr> 
   <td align="center">GEOMETRY</td> 
   <td align="right">String</td> 
  </tr> 
  <tr> 
   <td align="center">BINARY</td> 
   <td align="right">FixedString</td> 
  </tr> 
  <tr> 
   <td align="center">BIT</td> 
   <td align="right">UInt64</td> 
  </tr> 
  <tr> 
   <td align="center">SET</td> 
   <td align="right">UInt64</td> 
  </tr> 
 </tbody> 
</table>

#### （6）创建语句以及配置参数 ####

CREATE DATABASE [IF NOT EXISTS] db_name [ON CLUSTER cluster]
    ENGINE = MaterializedMySQL('host:port', ['database' | database], 'user', 'password') [SETTINGS ...]
    [TABLE OVERRIDE table1 (...), TABLE OVERRIDE table2 (...)]

引擎参数

*  `host:port`— MySQL 服务器端点。
 *  `database`— MySQL 数据库名称。
 *  `user`— MySQL 用户。
 *  `password`\- 用户密码。

引擎设置

*  `max_rows_in_buffer`— 允许数据在内存中缓存的最大行数（对于单个表且缓存数据无法查询）。当超过这个数字时，数据将被物化。默认值：65505。
 *  `max_bytes_in_buffer`— 允许数据在内存中缓存的最大字节数（对于单个表且缓存数据无法查询）。当超过这个数字时，数据将被物化。默认值：1048576。
 *  `max_rows_in_buffers`— 允许数据在内存中缓存的最大行数（对于数据库和缓存数据无法查询）。当超过这个数字时，数据将被物化。默认值：65505。
 *  `max_bytes_in_buffers`— 允许数据在内存中缓存的最大字节数（对于数据库和缓存数据无法查询）。当超过这个数字时，数据将被物化。默认值：1048576。
 *  `max_flush_data_time`— 允许数据在内存中缓存的最大毫秒数（对于数据库和缓存数据无法查询）。当超过这个时间时，数据将被物化。默认值：1000。
 *  `max_wait_time_when_mysql_unavailable`— MySQL 不可用时的重试间隔（毫秒）。负值禁用重试。默认值：1000。
 *  `allows_query_when_mysql_lost`— 允许在 MySQL 丢失时查询物化表。默认值：( 0) false。
 *  `materialized_mysql_tables_list`— 以逗号分隔的 mysql 数据库表列表，将由 MaterializedMySQL 数据库引擎复制。默认值：空列表 — 表示将复制整个表。

#### （7）表覆盖 ####

表覆盖可用于自定义 ClickHouse DDL 查询，允许您为应用程序进行架构优化。这对于控制分区特别有用，这对 MaterializedMySQL 的整体性能很重要。

这些是您可以对 MaterializedMySQL 的表覆盖进行的模式转换操作：

*  修改列类型。必须与原始类型兼容，否则复制将失败。例如，您可以将 UInt32 列修改为 UInt64，但不能将 String 列修改为 Array(String)。
 *  修改列 TTL。
 *  修改列压缩编解码器。
 *  添加别名列。
 *  添加跳过索引
 *  添加投影。请注意，使用时会禁用投影优化SELECT … FINAL（MaterializedMySQL 默认会这样做），因此它们的实用性在这里受到限制。
 *  修改PARTITION BY
 *  修改ORDER BY
 *  修改主键
 *  添加样品
 *  添加表 TTL

CREATE DATABASE db_name ENGINE = MaterializedMySQL(...)
    [SETTINGS ...]
    [TABLE OVERRIDE table_name (
        [COLUMNS (
            [col_name [datatype] [ALIAS expr] [CODEC(...)] [TTL expr], ...]
            [INDEX index_name expr TYPE indextype[(...)] GRANULARITY val, ...]
            [PROJECTION projection_name (SELECT <COLUMN LIST EXPR> [GROUP BY] [ORDER BY]), ...]
        )]
        [ORDER BY expr]
        [PRIMARY KEY expr]
        [PARTITION BY expr]
        [SAMPLE BY expr]
        [TTL expr]
    ), ...]

示例：

CREATE DATABASE db_name ENGINE = MaterializedMySQL(...)
    TABLE OVERRIDE table1 (
        COLUMNS (
            userid UUID,
            category LowCardinality(String),
            timestamp DateTime CODEC(Delta, Default)
        )
        PARTITION BY toYear(timestamp)
    ),
    TABLE OVERRIDE table2 (
        COLUMNS (
            client_ip String TTL created + INTERVAL 72 HOUR
        )
        SAMPLE BY ip_hash
    )

`COLUMNS`名单很稀疏；根据指定修改现有列，添加额外的 ALIAS 列。无法添加普通或 MATERIALIZED 列。具有不同类型的修改列必须可以从原始类型分配。目前在`CREATE DATABASE`执行查询时没有验证此问题或类似问题，因此需要格外小心。

可以为尚不存在的表指定覆盖。

**！！！警告如果不小心使用，很容易使用表覆盖来破坏复制。**

#### （8）注意事项 ####

①MySQL中的每个表都应包含PRIMARY KEY.

②表的复制，那些包含ENUM字段值超出范围（在ENUM签名中指定）的行将不起作用。

③如果 ClickHouse 无法解析某些 DDL 查询，则忽略该查询。

## 二.案例实操 ##

### 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 即可。

查询以下语句进行验证：

show variables like '%gtid_mode%';
    show variables like '%enforce_gtid_consistency%';
    show variables like '%binlog_format%';

gtid_mode    ON
    enforce_gtid_consistency    ON
    binlog_format    ROW

如以上值，则代表可以进行同步。

#### （3）重启 MySQL ####

sudo systemctl restart mysqld

#### （4）用户 ####

同步用户建议最高级用户（拥有all权限）或者用户有RELOAD, REPLICATION SLAVE, REPLICATION CLIENT相关权限。

如为普通用户【5.7版本mysql】需执行 \{必须 *.* 才是服务服务器权限\}

grant select,reload,replication slave,replication client on *.* to syncdata@'%' identified  by "123";
    flush privileges;--刷新权限

### 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);

### 3. 开启 ClickHouse 物化引擎 ###

注意：以下语句如果运行设置找不到，需要先在users.xml开启调用窗口函数使用状态在/etc/clickhouse-server/users.xml 配置文件中加入<allow\_experimental\_window\_functions>1</allow\_experimental\_window\_functions>

<?xml version="1.0"?>
    <yandex>
    
    <profiles>
    
    <default>
    
    <max_memory_usage>10000000000</max_memory_usage>
     
    
    <load_balancing>random</load_balancing>
    <allow_experimental_window_functions>1</allow_experimental_window_functions>
    </default>
     
    
    <readonly>
    <readonly>1</readonly>
    </readonly>
    </profiles>
    ......

然后开启 MaterializeMySQL 库引擎操作权限，然后配置文件中的allow\_experimental\_window\_functions 可以去除【高版本会提示过时，旧版本可能需要保留，看登录的提示信息】

注意每次登录默认是关闭的，需要使用再操作打开。

set allow_experimental_database_materialize_mysql=1;

### 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;

### 5. 修改数据 ###

#### （1）在 MySQL 中修改数据: ####

update t_organization set name = CONCAT(name,'-v1') where id = 1

#### （2）查看 clickhouse 日志可以看到 binlog 监听事件，查询 clickhouse ####

select * from t_organization;

### 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

### 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;