ClickHouse数据一致性

た入场券 2024-03-23 16:07 48阅读 0赞

#### 目录 ####

*  1 准备测试表和数据
 *  2 手动OPTIMIZE(不推荐)
 *  3 通过 Group by 去重
 *  4 通过 FINAL 查询
 *   *  4.1 老版本测试
     *  4.2 新版本测试

--------------------

## 1 准备测试表和数据 ##

![在这里插入图片描述][ee43ce5ccfb348789474813f64ae2445.png]

查询 CK 手册发现，即便对数据一致性支持最好的 Mergetree，也只是`保证最终一致性`：  
![在这里插入图片描述][db7ba616831648909daa18d50ff5c5cf.png]

我们在使用 ReplacingMergeTree、SummingMergeTree 这类表引擎的时候，会出现短暂数据不一致的情况。  
  在某些对一致性非常敏感的场景，通常有以下几种解决方案。

1.  创建表

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 代表删除数据。

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

1.  修改前 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);

1.  统计总数

SELECT COUNT() FROM test_a;
       10500000
       12

还未触发分区合并，所以还未去重。

## 2 手动OPTIMIZE(不推荐) ##

在写入数据后，立刻执行 OPTIMIZE 强制触发新写入分区的合并动作。

OPTIMIZE TABLE test_a FINAL;
    
    语法：OPTIMIZE TABLE [db.]name [ON CLUSTER cluster] [PARTITION partition | 
    PARTITION ID 'partition_id'] [FINAL] [DEDUPLICATE [BY expression]]
    1234

## 3 通过 Group by 去重 ##

1.  执行去重的查询

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\_time 得到修改后的 score 字段值。

1.  创建视图，方便测试

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;

1.  插入重复数据，再次查询

#再次插入一条数据
       INSERT INTO TABLE test_a(user_id,score,create_time)
       VALUES(0,'AAAA',now())
       #再次查询
       SELECT *
       FROM view_test_a
       WHERE user_id = 0;

1.  删除数据测试

#再次插入一条标记为删除的数据
       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 最终将物理数据删除。

## 4 通过 FINAL 查询 ##

在查询语句后增加 FINAL 修饰符，这样在查询的过程中将会执行 Merge 的特殊逻辑（例如数据去重，预聚合等）。  
  但是这种方法在早期版本基本没有人使用，因为在增加 FINAL 之后，我们的查询将会变成一个单线程的执行过程，查询速度非常慢。  
  在 v20.5.2.7-stable 版本中，FINAL 查询支持多线程执行，并且可以通过 max\_final\_threads 参数控制单个查询的线程数。但是目前读取 part 部分的动作依然是串行的。  
  FINAL 查询最终的性能和很多因素相关，列字段的大小、分区的数量等等都会影响到最终的查询时间，所以还要结合实际场景取舍。

参考链接：https://github.com/ClickHouse/ClickHouse/pull/10463  
分别安装了 20.4.5.36 和 21.7.3.14 两个版本的 ClickHouse 进行对比。

### 4.1 老版本测试 ###

1.  普通查询语句

select * from visits_v1 WHERE StartDate = '2014-03-17' limit 100;

1.  FINAL 查询

select * from visits_v1 FINAL WHERE StartDate = '2014-03-17' limit 100;

先前的并行查询变成了单线程。

### 4.2 新版本测试 ###

1.  普通语句查询

select * from visits_v1 WHERE StartDate = '2014-03-17' limit 100 settings max_threads = 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

明显将由 2 个线程并行读取 part 查询。

1.  FINAL 查询

select * from visits_v1 final WHERE StartDate = '2014-03-17' limit 100 
       settings max_final_threads = 2;

查询速度没有普通的查询快，但是相比之前已经有了一些提升,查看 FINAL 查询的执行计划：

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

从 CollapsingSortedTransform 这一步开始已经是多线程执行，但是读取 part 部分的动作还是串行。

[ee43ce5ccfb348789474813f64ae2445.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/aa403dab18934eaa812f598319ad2609.png
[db7ba616831648909daa18d50ff5c5cf.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/00b892e0bc7a411b97fa5e871795c23e.png