ClickHouse实战--使用ReplacingMergeTree来更新数据

ClickHouse实战–使用ReplacingMergeTree来更新数据

概述

通过ReplacingMergeTree引擎的表可以更新主键相同和版本相同的数据行。本文介绍ReplacingMergeTree引擎的使用方式和数据更新的操作。

ReplacingMergeTree实战

创建本地表

创建一张本地表：stu_local

CREATE TABLE if not exists test_db.stu_local on cluster perftest_2shards_1replicas
(`id` Int64,`stu_code` Int32,`stu_name` String,`total_cnt` Int64,`event_time` DateTime,`version`  DateTime
)
Engine=ReplacingMergeTree(version)
PARTITION BY toYYYYMMDD(event_time)
ORDER BY id;

注意：on cluster后面的是集群名，这个根据system.clusters表中的名称进行修改。

创建分布式表

CREATE TABLE if not exists test_db.stu_all on cluster perftest_2shards_1replicas
as test_db.stu_local
ENGINE = Distributed(perftest_2shards_1replicas, test_db, stu_local, intHash64(id));

注意：为了让相同id的数据行分布到同一个节点上，就不能再使用rand()这个sharding策略。而是需要通过一个hash函数来让相同主键的数据分片到相同的节点上。

由于，我们的主键是：id。所以，可以使用一个hash函数来做这个分片策略。hash函数的说明可以参考官方文档：Hash Functions。

向分布式表插入数据

insert into test_db.stu_all (id, stu_code, stu_name, total_cnt, event_time,version) values (1, 4000, 'guangzhou', 420000, toDateTime(now()), toDateTime(now()));
insert into test_db.stu_all (id, stu_code, stu_name, total_cnt, event_time,version) values (2, 5000, 'shenzhan', 55000, toDateTime(now()), toDateTime(now()));
insert into test_db.stu_all (id, stu_code, stu_name, total_cnt, event_time,version) values (3, 6000, 'huizhou', 65000, toDateTime(now()), toDateTime(now()));
insert into test_db.stu_all (id, stu_code, stu_name, total_cnt, event_time,version) values (4, 7000, 'huizhou', 75000, toDateTime(now()), toDateTime(now()));
insert into test_db.stu_all (id, stu_code, stu_name, total_cnt, event_time,version) values (5, 8000, 'huizhou', 75001, toDateTime(now()), toDateTime(now()));

再插入两条相同的id为1和2的数据，此时version的值会获取当前的时间，所以和以前的数据不同。

insert into test_db.stu_all (id, stu_code, stu_name, total_cnt, event_time,version) values (1, 4000, 'guangzhou', 420000, toDateTime(now()), toDateTime(now()));
insert into test_db.stu_all (id, stu_code, stu_name, total_cnt, event_time,version) values (2, 5000, 'shenzhan', 55000, toDateTime(now()), toDateTime(now()));

此时可以看到，表中的数据如下：

id   stu_code    stu_name    total_cnt   event_time  version
2   5000    shenzhan    55000   2022-05-03 16:44:19.000 2022-05-03 16:44:19.000
2   5000    shenzhan    55000   2022-05-03 16:50:01.000 2022-05-03 16:50:01.000
1   4000    guangzhou   420000  2022-05-03 16:50:01.000 2022-05-03 16:50:01.000
1   4000    guangzhou   420000  2022-05-03 16:45:53.000 2022-05-03 16:45:53.000
3   6000    huizhou 65000   2022-05-03 16:48:19.000 2022-05-03 16:48:19.000
4   7000    huizhou 75000   2022-05-03 16:48:28.000 2022-05-03 16:48:28.000
5   8000    huizhou 75001   2022-05-03 16:48:34.000 2022-05-03 16:48:34.000

添加一个final字段后的结果如下：

SELECT * from test_db.stu_all sa final;

该sql的执行结果如下：

id   stu_code    stu_name    total_cnt   event_time  version
2   5000    shenzhan    55000   2022-05-03 16:50:01.000 2022-05-03 16:50:01.000
1   4000    guangzhou   420000  2022-05-03 16:50:01.000 2022-05-03 16:50:01.000
4   7000    huizhou 75000   2022-05-03 16:48:28.000 2022-05-03 16:48:28.000
5   8000    huizhou 75001   2022-05-03 16:48:34.000 2022-05-03 16:48:34.000
3   6000    huizhou 65000   2022-05-03 16:48:19.000 2022-05-03 16:48:19.000

OPTIMIZE语句

可以使用OPTIMIZE语句来让数据合并立刻生效。但该语句会引发大量的读写，所以，需要谨慎使用。

OPTIMIZE table test_db.stu_local on cluster perftest_2shards_1replicas;

再次执行查询sql，可以看到数据已经被合并了。

以上例子的问题

注意以上例子可能达不到预期的效果，本来我们希望相同id的数据合并（只保留一条），但以上建表方式可能达不到效果。为什么呢？

其实，原因是由于我们的分区字段创建的有问题。我们说过，只有相同分区的数据才会合并，我们使用的是toYYYYMMDD(event_time)–按天的日期作为分区，这意味着，同一天的相同的id数据行会在同一个分区，会被合并。而不同日期的数据是不会被合并的。

插入两条不同日期的数据：

insert into test_db.stu_all (id, stu_code, stu_name, total_cnt, event_time,version) values (1, 4000, 'guangzhou', 420000, toDateTime('2022-02-21 00:00:00'), toDateTime(now()));insert into test_db.stu_all (id, stu_code, stu_name, total_cnt, event_time,version) values (2, 5000, 'shenzhan', 55000, toDateTime('2022-02-21 00:00:00'), toDateTime(now()));

执行optimize语句：

OPTIMIZE table test_db.stu_local on cluster perftest_2shards_1replicas;

再查询一下数据：

id   stu_code    stu_name    total_cnt   event_time  version1    4000    guangzhou   420000  2022-05-03 16:50:01.000 2022-05-03 16:50:01.0001    4000    guangzhou   420000  2022-02-21 00:00:00.000 2022-05-03 17:07:22.0002    5000    shenzhan    55000   2022-05-03 16:50:01.000 2022-05-03 16:50:01.0002    5000    shenzhan    55000   2022-02-21 00:00:00.000 2022-05-03 17:07:26.0003    6000    huizhou 65000   2022-05-03 16:48:19.000 2022-05-03 16:48:19.0004    7000    huizhou 75000   2022-05-03 16:48:28.000 2022-05-03 16:48:28.0005    8000    huizhou 75001   2022-05-03 16:48:34.000 2022-05-03 16:48:34.000

这说明了Clickhouse没有相同id且分区不同的数据行。

所以，要想实现数据合并，需要让相同id的数据落到相同分区中。比如对于用户信息，可以添加一个部门等字段，那这个字段作为分区，这样就可以让相同用户id的数据落到相同分区了。

但注意，虽然后台并没有实现数据合并，但当我们添加final时，查询的结果却是正确的：

SELECT * from test_db.stu_all sa final order by id;

通过添加final能得到正确的结果。

id   stu_code    stu_name    total_cnt   event_time  version1    4000    guangzhou   420000  2022-02-21 00:00:00.000 2022-05-03 17:07:22.0002    5000    shenzhan    55000   2022-02-21 00:00:00.000 2022-05-03 17:07:26.0003    6000    huizhou 65000   2022-05-03 16:48:19.000 2022-05-03 16:48:19.0004    7000    huizhou 75000   2022-05-03 16:48:28.000 2022-05-03 16:48:28.0005    8000    huizhou 75001   2022-05-03 16:48:34.000 2022-05-03 16:48:34.000

但由于数据并没有真正合并，所以添加final查询时，随着垃圾数据越来越多，会严重影响性能。所以，还是需要想办法把该删除的数据合并掉。

小结

本文通过一些例子介绍了如何使用ReplacingMergeTree引擎。在使用ReplacingMergeTree要注意：（1）只有相同分区的数据行才会被合并，所以，必须要注意让需要合并的数据落到相同分区中。（2）若是分布式表，还需要注意要让需要合并的数据落到相同的节点上。（3）不能频繁使用optimize操作；（4）ReplacingMergeTree有一定的适用场景，不是所有的场景，需要根据业务需要进行选择。