ClickHouse与MySQL之间实时同步数据（MySQL引擎），将MySQL数据实时同步到clickhouse

Mysql 部分配置

开启binlog并设置为row格式:
在MySQL配置文件/etc/my.cnf中加入

server-id=1 
log-bin=mysqlbin.log
binlog_format=ROW

开启GTID模式(在MySQL配置文件/etc/my.cnf中加入)

gtid-mode=on
enforce-gtid-consistency=1  # 设置为主从强一致性
log-slave-updates=1 # 记录日志

查询以下语句进行验证

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

default_authentication_plugin 参数的设定，在MySQL配置文件/etc/my.cnf中加入

查看现有配置
show variables like 'default_authentication_plugin';

加入如下配置：

	default_authentication_plugin=mysql_native_password

注意使用的用户密码插件必须为 mysql_native_password，否则会报如下错误：

详情请看：https://github.com/ClickHouse/ClickHouse/issues/28514 ，出现这种错误，创建新用户用 mysql_native_password ，并有root权限，远程访问，如：

create user 'copy_root'@'%' identified by '12345678';
grant all privileges on *.* to 'copy_root'@'%';

grant system_user on *.* to 'root';
alter user 'copy_root'@'%' identified with mysql_native_password by '12345678';
flush privileges;

将mysql的copy_root权限设置为允许所有ip地址可以访问，执行如下sql语句

update mysql.user set host='%' where user='copy_root';  
FLUSH PRIVILEGES;     --刷新权限

注意：这里用户(copy_root)必须要有：FLUSH· TABLES 权限，否则会报如下错误：
在这里插入图片描述
7. 创建测试库(准备 MySQL 表和数据)

CREATE DATABASE IF NOT EXISTS test

clickhouse 配置

连接数据库,如图

输入clickhouse-client，输入密码

clickhouse-client;

开启同步：执行如下命令

set allow_experimental_database_materialized_mysql=1;

同步语法说明：

CREATE DATABASE [IF NOT EXISTS] db_name [ON CLUSTER cluster]
ENGINE = MySQL('host:port', ['database' | database], 'user', 'password')

同步 ( test ) 库下所有表，命令如下：

CREATE DATABASE IF NOT EXISTS test ENGINE = MaterializedMySQL('47.132.151.48:3306', 'test', 'copy_root', '12345678');

同步 ( test )库下指定表 ( demo )，命令如下：

CREATE DATABASE IF NOT EXISTS test ENGINE = MaterializedMySQL('47.132.151.48:3306', 'test', 'copy_root', '12345678')
SETTINGS materialized_mysql_tables_list = 'demo';

interserver_http_host参数设置：

该配置防止报如下错误：

DB::StorageReplicatedMergeTree::processQueueEntry(ReplicatedMergeTreeQueue::SelectedEntryPtr)::(anonymous class)::operator()(DB::StorageReplicatedMergeTree::LogEntryPtr &) const: Poco::Exception. Code: 1000, e.code() = 111, Connection refused

创建表测试