ClickHouse 的 ReplacingMergeTree 等引擎只能保证数据的最终一致性,在查询时可能会出现短暂数据不一致的情况(有些企业业务为了性能可以容忍一些小的不一致)
一、数据准备
--建表
CREATE TABLE test_a(
user_id UInt64,
score String,
deleted UInt8 DEFAULT 0,
create_time DateTime DEFAULT toDateTime(0)
)ENGINE= ReplacingMergeTree(create_time)
ORDER BY user_id;
--user_id 是数据去重更新的标识
--create_time 是版本号字段,每组数据中 create_time 最大的一行表示最新的数据
--deleted 是自定的一个标记位,比如 0 代表未删除,1 代表删除数据
--插入1000万测试数据
INSERT INTO TABLE test_a(user_id, score)
WITH(
SELECT ['A','B','C','D','E','F','G']
) AS dict
SELECT number AS user_id, dict[number%7+1] FROM numbers(10000000);
--修改前 50 万 行数据,修改内容包括 name 字段和 create_time 版本号字段
INSERT INTO TABLE test_a(user_id, score, create_time)
WITH(
SELECT ['AA','BB','CC','DD','EE','FF','GG']
)AS dict
SELECT number AS user_id, dict[number%7+1], now() AS create_time FROM
numbers(500000);
--查询数据条数
SELECT COUNT() FROM test_a; --存在重复数据
二、手动 optimize
生产上不建议
--语法:OPTIMIZE TABLE [db.]name [ON CLUSTER cluster] [PARTITION partition | PARTITION ID 'partition_id'] [FINAL] [DEDUPLICATE [BY expression]]
OPTIMIZE TABLE test_a FINAL;
SELECT COUNT() FROM test_a; --重复数据去除了
三、通过 group by 去重
--使用 group by 语法进行去重查询
SELECT
user_id ,
argMax(score, create_time) AS score,
argMax(deleted, create_time) AS deleted,
max(create_time) AS ctime
FROM test_a
GROUP BY user_id
HAVING deleted = 0;
--argMax(field1, field2):取 field2 最大值的那条数据的 field1 字段值
--把上述语句创建为视图,方便测试
CREATE VIEW view_test_a AS
SELECT
user_id ,
argMax(score, create_time) AS score,
argMax(deleted, create_time) AS deleted,
max(create_time) AS ctime
FROM test_a
GROUP BY user_id
HAVING deleted = 0;
--插入重复数据测试
INSERT INTO TABLE test_a(user_id,score,create_time) VALUES(0,'AAAA',now());
SELECT * FROM view_test_a WHERE user_id = 0; --返回最新插入的数据
--删除数据测试
INSERT INTO TABLE test_a(user_id,score,deleted,create_time) VALUES(0,'AAAA',1,now());
SELECT * FROM view_test_a WHERE user_id = 0; --无数据
--最终可以结合表级别的 TTL 将物理数据真正删除
四、通过 final 查询
在查询语句后增加 final 修饰符,这样在查询的过程中将会执行 Merge 的特殊逻辑(例如数据去重,预聚合等),但在 20.5 及以后的版本 final 才支持多线程
1. 老版本测试
select * from visits_v1 WHERE StartDate = '2014-03-17' limit 100;
select * from visits_v1 final WHERE StartDate = '2014-03-17' limit 100;
2. 新版本测试
explain pipeline select * from visits_v1 WHERE StartDate = '2014-03-17' limit 100 settings max_threads = 2;
(Expression)
ExpressionTransform × 2
(SettingQuotaAndLimits)
(Limit)
Limit 2 → 2
(ReadFromMergeTree)
MergeTreeThread × 2 0 → 1
explain pipeline select * from visits_v1 final WHERE StartDate = '2014-03-17' limit 100 settings max_final_threads = 2;
(Expression)
ExpressionTransform × 2
(SettingQuotaAndLimits)
(Limit)
Limit 2 → 2
(ReadFromMergeTree)
ExpressionTransform × 2
CollapsingSortedTransform × 2
Copy 1 → 2
AddingSelector
ExpressionTransform
MergeTree 0 → 1 --但是读取 part 部分的动作还是串行