高性价比OLAP列式数据库—Clickhouse（数据一致性问题）

迷三张 2025-02-24

135

相信很多使用了ClickHouse数据库的小伙伴在开发工作中基本都遇到了库中数据有很多条重复的情况，且是偶发出现，这就令人费解了不是

，下面我们一起去解密~~~

数据一致性是什么？

用大白话来讲就是在分布式系统或者数据库中，数据是存储到不同节点的。而数据一致性呢其实就是一种状态，保证我们读取数据的时候在任何时候都是一致的，相同的。

提到它小伙伴估计会想起另外两个，数据可用性，数据分区容错性。三者共同为分布式三大核心特征。

一致性分为四个Level等级，强一致性、最终一致性、弱一致性、会话一致性，四者都是有利也有弊。

而最终一致性优点就是不会给分布式系统性很大节点间读写数据合并压力，所以不会影响性能；缺点就是在某些时候，系统可能会读到过时或不一致的数据。

ClickHouse的数据一致性

数据一致性其实小伙伴们本不陌生的，无论是在开发中高并发场景，还是开发使用的中间件、工具，以及分布式架构下的系统中它永远都是主要挑战之一。而ClickHouse属于AP类型的数据库，又常常用于大数据场景下，要保证高吞吐高性能，所以一致性问题它必须要直面。

合并树引擎【MergeTree】家族下的成员【ReplacingMergeTree】能够对数据去重，但也只是保证最终一致性。如下，官网说：

解决方案

首先准备测试数据

创建表模型

CREATE TABLE test_z(
 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_z(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万数据

INSERT INTO TABLE test_z(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_z;
10500000

是不是很惊讶为什么1050万数据，因为还未触发分区合并，所以还没有去重。

方案一：手动执行OPTIMIZE

语法

OPTIMIZE TABLE [db.]name [ON CLUSTER cluster] [PARTITION partition | PARTITION ID 'partition_id'] [FINAL | FORCE] [DEDUPLICATE [BY expression]]

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

OPTIMIZE TABLE test_z FINAL;

但是数据合并会在后台一个不确定的时间进行，因此你是无法预先作出计划的，不可能每次都人工的去操作执行下。

所以你需要在开发中设置一个拦截器或者切面的方式去管理有数据更新操作的接口，每次调用都在响应后自动执行OPTIMIZE操作；

当然如果你使用的ORM框架为MyBatis系列的，可以自定义MyBatis的拦截器实现。

@Intercepts({
    @org.apache.ibatis.plugin.Signature(type = StatementHandler.class, method = "query", args = { Statement.class })
})
public class MyCustomInterceptor implements org.apache.ibatis.plugin.Interceptor {
    @Override
    public Object intercept(Invocation invocation) throws Throwable {
        / 执行查询逻辑
        Object result = invocation.proceed(); 
        / 在查询响应后执行你自定义的逻辑
        / 例如：你可以在这里执行一些额外的 SQL 语句
        System.out.println("OPTIMIZE TABLE test_z FINAL;");
        / 在这里你可以选择执行一些 SQL，例如更新日志、统计等
        return result; 
    }
    @Override
    public Object plugin(Object target) {
        return Plugin.wrap(target, this);
    }
}

将自定义的拦截器加入到 MybatisPlusInterceptor
中

@Bean
public MybatisPlusInterceptor mybatisPlusInterceptor() {


    MybatisPlusInterceptor interceptor = new MybatisPlusInterceptor();
    // 添加自定义拦截器


    interceptor.addInnerInterceptor(new MyCustomInterceptor());
    return interceptor;
}

方案二：通过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_time从而得到其他字段的最新值，而这样做只是把重复的数据过滤了，可以结合TTL机制把数据进行物理删除。这个函数在工作中很常用，建议深刻记忆~~~

方案三：通过FINAL修饰符查询

在查询语句后增加 FINAL 修饰符，这样在查询的过程中将会执行 Merge 的特殊逻辑（例如数据去重，预聚合等），它会强制 ClickHouse 在查询时对数据进行最终的合并，以确保返回的结果是最终一致的。

一般涉及到具有强制性的概念，我们都知道肯定要牺牲一些东西去换想要的效果。所以使用FINAL修饰词的代价就是会导致查询性能下降，因为它需要在查询时对所有数据进行额外的排序和合并操作。这种操作通常是单线程的，尤其是在处理大规模数据时，性能可能会成为瓶颈。

ClickHouse在20.x版本前对FINAL修饰词支持不好，几乎没人愿意去用。但在21.x版本后对其进行了优化，查询底层变为了串行，虽然不是多线程执行但对查询性能提升了很大。

测试对比如下：

普通查询

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

explain执行计划

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 个线程并行读取的查询。

FINAL查询

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

查询速度没有普通的查询快，但是相比之前老版本已经有了很多提升。

explain执行计划

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 这一步开始已经是多线程执行，但是读取具体数据部分的动作还是串行。

总结

目前我使用版本23.3.9.55，如果在开发中需要使用到FINAL，我建议限制语句查询范围（如时间范围、过滤条件），减少需要处理的数据量。
在项目木尽量避免全面使用FINAL，只对数据幂等性要求高的功能点进行使用。
FINAL 以及语句Group By去重和预合并数据（OPTIMIZE
）结合使用效果最好。

clickhouse 大数据分布式一致性数据库一致性

文章转载自迷三张，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

高性价比OLAP列式数据库—Clickhouse（数据一致性问题）

评论