Last updated: 2020-09-08

Many people might think they clicked on the wrong link seeing this title, but no.

ClickHouse can be mounted as a MySQL replica, performing full-then-incremental real-time synchronization of MySQL data. This feature can be described as this year’s most eye-catching and most-needed functionality. Based on it, we can easily build enterprise-level solutions, making OLTP and OLAP integration no longer a headache.

Currently supports MySQL 5.6/5.7/8.0 versions, compatible with Delete/Update statements and most common DDL operations.
The code has been merged into the upstream master branch and is expected to be released as an experimental feature in version 20.8.

After all, this is the integration of two heterogeneous ecosystems. There’s still considerable work to be done, and we look forward to community user feedback to accelerate iteration.

Code Acquisition

Get clickhouse/master code and compile. Method described in ClickHouse and Friends (1) Compilation, Development, Testing…

MySQL Master

We need a MySQL with binlog enabled as master:

1
docker run -d -e MYSQL_ROOT_PASSWORD=123 mysql:5.7 mysqld --datadir=/var/lib/mysql --server-id=1 --log-bin=/var/lib/mysql/mysql-bin.log --gtid-mode=ON --enforce-gtid-consistency

Create database and table, then insert data:

1
2
3
4
5
6
7
8
9
10
11
12
mysql> create database ckdb;
mysql> use ckdb;
mysql> create table t1(a int not null primary key, b int);
mysql> insert into t1 values(1,1),(2,2);
mysql> select * from t1;
+---+------+
| a | b    |
+---+------+
| 1 |    1 |
| 2 |    2 |
+---+------+
2 rows in set (0.00 sec)

ClickHouse Slave

Currently replication is done at the database level. Different databases can come from different MySQL masters, enabling synchronization of data from multiple MySQL sources to one ClickHouse for OLAP analysis.

First enable the experimental feature flag:

1
clickhouse :) SET allow_experimental_database_materialize_mysql=1;

Create a replication channel:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
clickhouse :) CREATE DATABASE ckdb ENGINE = MaterializeMySQL('172.17.0.2:3306', 'ckdb', 'root', '123');
clickhouse :) use ckdb;
clickhouse :) show tables;
┌─name─┐
│ t1   │
└──────┘
clickhouse :) select * from t1;
┌─a─┬─b─┐
│ 1 │ 1 │
└───┴───┘
┌─a─┬─b─┐
│ 2 │ 2 │
└───┴───┘

2 rows in set. Elapsed: 0.017 sec.

Check ClickHouse’s synchronization position:
cat ckdatas/metadata/ckdb/.metadata

1
2
3
4
Version:	1
Binlog File:	mysql-bin.000001
Binlog Position:	913
Data Version:	0

Delete

First execute a delete operation on MySQL Master:

1
2
mysql> delete from t1 where a=1;
Query OK, 1 row affected (0.01 sec)

Then check records on ClickHouse Slave:

1
2
3
4
5
6
7
8
9
10
clickhouse :) select * from t1;

SELECT *
FROM t1

┌─a─┬─b─┐
│ 2 │ 2 │
└───┴───┘

1 rows in set. Elapsed: 0.032 sec.

At this point, Data Version in metadata has incremented to 2:

1
2
3
4
5
cat ckdatas/metadata/ckdb/.metadata
Version:	1
Binlog File:	mysql-bin.000001
Binlog Position:	1171
Data Version:	2

Update

MySQL Master:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
mysql> select * from t1;
+---+------+
| a | b    |
+---+------+
| 2 |    2 |
+---+------+
1 row in set (0.00 sec)

mysql> update t1 set b=b+1;

mysql> select * from t1;
+---+------+
| a | b    |
+---+------+
| 2 |    3 |
+---+------+
1 row in set (0.00 sec)

ClickHouse Slave:

1
2
3
4
5
6
7
8
9
10
clickhouse :) select * from t1;

SELECT *
FROM t1

┌─a─┬─b─┐
│ 2 │ 3 │
└───┴───┘

1 rows in set. Elapsed: 0.023 sec.

Performance Testing

Test Environment

1
2
3
MySQL          8C16G cloud VM, 192.168.0.3, 10188183 records baseline
ClickHouse     8C16G cloud VM, 192.168.0.4
benchyou       8C8G cloud VM, 192.168.0.5, 256 concurrent writes, https://github.com/xelabs/benchyou

Performance testing is heavily related to hardware environment. Cloud VMs are used here, data for reference only.

Full Synchronization Performance

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
8c16G-vm :) create database sbtest engine=MaterializeMySQL('192.168.0.3:3306', 'sbtest', 'test', '123');

8c16G-vm :) watch lv1;

WATCH lv1

┌─count()─┬───────────────now()─┬─_version─┐
│       0 │ 2020-07-29 06:36:04 │        1 │
└─────────┴─────────────────────┴──────────┘
┌─count()─┬───────────────now()─┬─_version─┐
│ 1113585 │ 2020-07-29 06:36:05 │        2 │
└─────────┴─────────────────────┴──────────┘
┌─count()─┬───────────────now()─┬─_version─┐
│ 2227170 │ 2020-07-29 06:36:07 │        3 │
└─────────┴─────────────────────┴──────────┘
┌─count()─┬───────────────now()─┬─_version─┐
│ 3340755 │ 2020-07-29 06:36:10 │        4 │
└─────────┴─────────────────────┴──────────┘
┌─count()─┬───────────────now()─┬─_version─┐
│ 4454340 │ 2020-07-29 06:36:13 │        5 │
└─────────┴─────────────────────┴──────────┘
┌─count()─┬───────────────now()─┬─_version─┐
│ 5567925 │ 2020-07-29 06:36:16 │        6 │
└─────────┴─────────────────────┴──────────┘
┌─count()─┬───────────────now()─┬─_version─┐
│ 6681510 │ 2020-07-29 06:36:18 │        7 │
└─────────┴─────────────────────┴──────────┘
┌─count()─┬───────────────now()─┬─_version─┐
│ 7795095 │ 2020-07-29 06:36:22 │        8 │
└─────────┴─────────────────────┴──────────┘
┌─count()─┬───────────────now()─┬─_version─┐
│ 8908680 │ 2020-07-29 06:36:25 │        9 │
└─────────┴─────────────────────┴──────────┘
┌──count()─┬───────────────now()─┬─_version─┐
│ 10022265 │ 2020-07-29 06:36:28 │       10 │
└──────────┴─────────────────────┴──────────┘
┌──count()─┬───────────────now()─┬─_version─┐
│ 10188183 │ 2020-07-29 06:36:28 │       11 │
└──────────┴─────────────────────┴──────────┘
← Progress: 11.00 rows, 220.00 B (0.16 rows/s., 3.17 B/s.)

In this hardware environment, full synchronization performance is approximately 424507/s, 420k transactions per second.
Since full data has no dependencies between rows, it can be further optimized for parallel processing to accelerate synchronization.
Full synchronization performance directly determines ClickHouse slave rebuild speed after failure. If your MySQL has 1 billion records, it can be rebuilt in approximately 40 minutes.

Incremental Performance (Real-Time Synchronization)

Under current configuration, ClickHouse slave’s single-threaded replay consumption capability exceeds MySQL master’s production capability under 256 concurrent connections. Testing shows they maintain real-time synchronization.

benchyou stress test data, 21k transactions/second (MySQL TPS doesn’t go higher in current environment):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
./bin/benchyou --mysql-host=192.168.0.3 --mysql-user=test --mysql-password=123 --oltp-tables-count=1 --write-threads=256 --read-threads=0

time            thds               tps     wtps    rtps
[13s]        [r:0,w:256,u:0,d:0]  19962    19962   0

time            thds               tps     wtps    rtps
[14s]        [r:0,w:256,u:0,d:0]  20415    20415   0

time            thds               tps     wtps    rtps
[15s]        [r:0,w:256,u:0,d:0]  21131    21131   0

time            thds               tps     wtps    rtps
[16s]        [r:0,w:256,u:0,d:0]  21606    21606   0

time            thds               tps     wtps    rtps
[17s]        [r:0,w:256,u:0,d:0]  22505    22505   0

ClickHouse side single-threaded replay capability, 21k transactions/second, real-time sync:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
┌─count()─┬───────────────now()─┬─_version─┐
│  150732 │ 2020-07-30 05:17:15 │       17 │
└─────────┴─────────────────────┴──────────┘
┌─count()─┬───────────────now()─┬─_version─┐
│  155477 │ 2020-07-30 05:17:16 │       18 │
└─────────┴─────────────────────┴──────────┘
┌─count()─┬───────────────now()─┬─_version─┐
│  160222 │ 2020-07-30 05:17:16 │       19 │
└─────────┴─────────────────────┴──────────┘
┌─count()─┬───────────────now()─┬─_version─┐
│  164967 │ 2020-07-30 05:17:16 │       20 │
└─────────┴─────────────────────┴──────────┘
┌─count()─┬───────────────now()─┬─_version─┐
│  169712 │ 2020-07-30 05:17:16 │       21 │
└─────────┴─────────────────────┴──────────┘
┌─count()─┬───────────────now()─┬─_version─┐
│  174457 │ 2020-07-30 05:17:16 │       22 │
└─────────┴─────────────────────┴──────────┘
┌─count()─┬───────────────now()─┬─_version─┐
│  179202 │ 2020-07-30 05:17:17 │       23 │
└─────────┴─────────────────────┴──────────┘
┌─count()─┬───────────────now()─┬─_version─┐
│  183947 │ 2020-07-30 05:17:17 │       24 │
└─────────┴─────────────────────┴──────────┘
┌─count()─┬───────────────now()─┬─_version─┐
│  188692 │ 2020-07-30 05:17:17 │       25 │
└─────────┴─────────────────────┴──────────┘
┌─count()─┬───────────────now()─┬─_version─┐
│  193437 │ 2020-07-30 05:17:17 │       26 │
└─────────┴─────────────────────┴──────────┘
┌─count()─┬───────────────now()─┬─_version─┐
│  198182 │ 2020-07-30 05:17:17 │       27 │
└─────────┴─────────────────────┴──────────┘

Implementation Mechanism

Before discussing the mechanism, we first need to understand MySQL binlog events. Main types include:

1
2
3
4
1. MYSQL_QUERY_EVENT       -- DDL
2. MYSQL_WRITE_ROWS_EVENT  -- insert data
3. MYSQL_UPDATE_ROWS_EVENT -- update data
4. MYSQL_DELETE_ROWS_EVENT -- delete data

When a transaction commits, MySQL processes executed SQL into corresponding binlog events and persists them to binlog files.

Binlog is MySQL’s important external output channel. As long as you implement MySQL Replication Protocol, you can stream-consume binlog events produced by MySQL. For specific protocol see Replication Protocol.

Due to historical reasons, the protocol is cumbersome and peculiar, which is not the focus of this article.

For ClickHouse consuming MySQL binlog, there are mainly 3 difficulties:

  • DDL compatibility
  • Delete/Update support
  • Query filtering

DDL

DDL compatibility consumed a large amount of code to implement.

First, let’s see what a MySQL table looks like after replicating to ClickHouse.

MySQL master:

1
2
3
4
5
6
7
8
mysql> show create table t1\G;
*************************** 1. row ***************************
       Table: t1
Create Table: CREATE TABLE `t1` (
  `a` int(11) NOT NULL,
  `b` int(11) DEFAULT NULL,
  PRIMARY KEY (`a`)
) ENGINE=InnoDB DEFAULT CHARSET=latin1

ClickHouse slave:

1
2
3
4
5
6
7
8
9
10
11
ATTACH TABLE t1
(
    `a` Int32,
    `b` Nullable(Int32),
    `_sign` Int8,
    `_version` UInt64
)
ENGINE = ReplacingMergeTree(_version)
PARTITION BY intDiv(a, 4294967)
ORDER BY tuple(a)
SETTINGS index_granularity = 8192

You can see:

  • Two hidden fields added by default: _sign(-1 delete, 1 insert) and _version(data version)
  • Engine converted to ReplacingMergeTree, using _version as column version
  • Original primary key field a serves as sorting and partitioning key

This is just one table’s replication. There are many other DDL operations to handle, such as adding columns, indexes, etc. Interested readers can review code under Parsers/MySQL.

Index Conversion

How do MySQL’s corresponding primary keys/indexes map to MaterializeMySQL table structure?

First scan keys in MySQL create table statement, InterpretersMySQLDDLQuery::getKeys:

  1. Scan unique_key
  2. Scan primary_key
  3. Scan auto_increment

Then generate OrderBy tuple expression based on keys in following order, InterpretersMySQLDDLQuery::getOrderByPolicy:

  1. primary_key[not increment]
  2. key[not increment]
  3. unique[not increment]
  4. unique[increment]
  5. key[increment]
  6. primary_key[increment]

ClickHouse currently has only one physical ordering (multiple physical orderings are somewhat planned), determined by OrderBy. So if MaterializeMySQL index is not well utilized, you can use materialized views to create new physical orderings.

Update and Delete

When we execute on MySQL master:

1
2
mysql> delete from t1 where a=1;
mysql> update t1 set b=b+1;

ClickHouse t1 data (querying _sign and _version together):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
clickhouse :) select a,b,_sign, _version from t1;

SELECT
    a,
    b,
    _sign,
    _version
FROM t1

┌─a─┬─b─┬─_sign─┬─_version─┐
│ 1 │ 1 │     1 │        1 │
│ 2 │ 2 │     1 │        1 │
└───┴───┴───────┴──────────┘
┌─a─┬─b─┬─_sign─┬─_version─┐
│ 1 │ 1 │    -1 │        2 │
└───┴───┴───────┴──────────┘
┌─a─┬─b─┬─_sign─┬─_version─┐
│ 2 │ 3 │     1 │        3 │
└───┴───┴───────┴──────────┘

According to returned results, you can see it’s composed of 3 parts.

part1 generated by mysql> insert into t1 values(1,1),(2,2):

1
2
3
4
┌─a─┬─b─┬─_sign─┬─_version─┐
│ 1 │ 1 │     1 │        1 │
│ 2 │ 2 │     1 │        1 │
└───┴───┴───────┴──────────┘

part2 generated by mysql> delete from t1 where a=1:

1
2
3
4
5
┌─a─┬─b─┬─_sign─┬─_version─┐
│ 1 │ 1 │    -1 │        2 │
└───┴───┴───────┴──────────┘
Note:
_sign = -1 indicates deleted state

part3 generated by update t1 set b=b+1:

1
2
3
┌─a─┬─b─┬─_sign─┬─_version─┐
│ 2 │ 3 │     1 │        3 │
└───┴───┴───────┴──────────┘

Query with final:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
clickhouse :) select a,b,_sign,_version from t1 final;

SELECT
    a,
    b,
    _sign,
    _version
FROM t1
FINAL

┌─a─┬─b─┬─_sign─┬─_version─┐
│ 1 │ 1 │    -1 │        2 │
└───┴───┴───────┴──────────┘
┌─a─┬─b─┬─_sign─┬─_version─┐
│ 2 │ 3 │     1 │        3 │
└───┴───┴───────┴──────────┘

2 rows in set. Elapsed: 0.016 sec.

You can see ReplacingMergeTree has deduplicated records based on _version and OrderBy.

Query

MySQL master:

1
2
3
4
5
6
7
mysql> select * from t1;
+---+------+
| a | b    |
+---+------+
| 2 |    3 |
+---+------+
1 row in set (0.00 sec)

ClickHouse slave:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
clickhouse :) select * from t1;

SELECT *
FROM t1

┌─a─┬─b─┐
│ 2 │ 3 │
└───┴───┘

clickhouse :) select *,_sign,_version from t1;

SELECT
    *,
    _sign,
    _version
FROM t1

┌─a─┬─b─┬─_sign─┬─_version─┐
│ 1 │ 1 │    -1 │        2 │
│ 2 │ 3 │     1 │        3 │
└───┴───┴───────┴──────────┘
Note: There's still a deleted record here, _sign is -1

MaterializeMySQL is defined as a storage engine, so during reads it judges based on _sign state. If -1, it’s deleted and gets filtered.

Parallel Replay

Why does MySQL need parallel replay?

Suppose MySQL master has 1024 concurrent writes/updates generating massive binlog events instantly. MySQL slave only has one thread replaying events one after another. So MySQL implemented parallel replay functionality!

So can MySQL slave replay completely (or nearly) simulate master’s 1024 concurrent behavior?

To parallelize, we first need to solve dependency issues: we need master to mark which events can be parallel and which have sequential relationships, since it’s the primary scene.

MySQL adds to binlog:

  • last_committed, same means can be parallel
  • sequence_number, smaller executes first, describes sequential dependencies
1
2
3
4
last_committed=3   sequence_number=4   -- event1
last_committed=4   sequence_number=5   -- event2
last_committed=4   sequence_number=6   -- event3
last_committed=5   sequence_number=7   -- event4

event2 and event3 can be parallel, event4 needs to wait for previous events to complete before replaying.
The above is just general principle. Currently MySQL has 3 parallel modes to choose from:

  1. Database-based parallelism
  2. Group commit-based parallelism
  3. Primary key conflict-free write set parallelism

Maximally accelerating MySQL slave replay, the entire mechanism is exceptionally complex.

Back to ClickHouse slave problem, we use single-threaded replay, lag is no longer the main issue, determined by their mechanisms:
MySQL slave replay needs to convert binlog events to SQL, then simulate master writes. This logical replication is the most important reason for low performance.
ClickHouse replay directly converts binlog events to underlying block structure, then directly writes to underlying storage engine, close to physical replication. Can be understood as replaying binlog events directly to InnoDB pages.

Read Latest

Although ClickHouse slave replay is very fast, near real-time, how to always read the latest data on ClickHouse slave?

Actually very simple. Leveraging MySQL binlog GTID feature, each read time we do an executed_gtid sync with master, then wait for these executed_gtid to finish replaying.

Data Consistency

For scenarios with high consistency requirements, how do we verify data consistency between MySQL master and ClickHouse slave?

Initial idea is to provide a function compatible with MySQL checksum algorithm. We just need to compare checksum values on both sides.

Summary

ClickHouse real-time replication syncing MySQL data is a 2020 upstream roadmap item. The overall architecture is quite challenging and no one took the ticket. Challenges mainly come from two aspects:

  • Very familiar with MySQL replication channels and protocols
  • Very familiar with overall ClickHouse mechanisms

Thus, a highway was built between two originally somewhat distant mountains. This 10851 highway was jointly constructed by two road workers: zhang1024 (ClickHouse side) and BohuTANG (MySQL replication), now merged into upstream branch.

Regarding syncing MySQL data, current solutions basically place a binlog consumption tool in the middle. This tool parses events, then converts to ClickHouse SQL statements, writing to ClickHouse server. Long chain, significant performance loss.

10851 highway implements a binlog consumption solution inside ClickHouse, then parses events into ClickHouse internal block structure, directly writing back to underlying storage engine. Almost the most efficient implementation method, achieving real-time MySQL synchronization capability, making analysis closer to reality.

Based on database-level replication, implementing multi-source replication functionality. If replication channel breaks, we just need to delete the database on ClickHouse side and recreate it. Very fast, convenient. OLTP+OLAP is that simple!

To get rich, first build roads!

【Top】ClickHouse MaterializeMySQL Real-Time MySQL Sync Summary