海量数据下的分库分表及ClickHouse解决方案

背景

最近在做的业务中，用户相关的数据不断增长，给系统带来了不小的压力，在 SQL优化实战-千万量级后的慢查 一文中也总结了一些针对慢查的解决方案。但每次活动下来，都会有几百上千万的用户相关数据产生，单纯的sql优化已经无法解决，本文站在前人肩膀上，总结了海量数据情景下的解决方案。

分区&分库分表

目前业务中使用的是MySQL，针对关系型数据库，可以采用分区或者分库分表的策略。首先看一下其各自的实现原理及优缺点：

（1）分区

• 分区原理：分区表是由多个相关的底层表实现，存储引擎管理分区的各个底层表和管理普通表一样，只是分区表在各个底层表上各自加上一个相同的索引（分区表要求所有的底层表都必须使用相同的存储引擎）。
• 分区优点：它对用户屏蔽了sharding的细节，即使查询条件没有sharding column，它也能正常工作（只是这时候性能一般）。
• 分区缺点：连接数、网络吞吐量等资源都受到单机的限制；并发能力远远达不到互联网高并发的要求。（主要因为虽然每个分区可以独立存储，但是分区表的总入口还是一个MySQL示例）。
• 适用场景：并发能力要求不高；数据不是海量（分区数有限，存储能力就有限）。

（2）分库分表

互联网行业处理海量数据的通用方法：分库分表。 分库分表中间件全部可以归结为两大类型：

• CLIENT模式；
• PROXY模式；

CLIENT模式代表有阿里的TDDL，开源社区的sharding-jdbc（sharding-jdbc的3.x版本即sharding-sphere已经支持了proxy模式）。架构如下：

PROXY模式代表有阿里的cobar，民间组织的MyCAT。架构如下：

无论是CLIENT模式，还是PROXY模式。几个核心的步骤是一样的：SQL解析，重写，路由，执行，结果归并。

分库分表实现（MYSQL）

针对分区与分库分表的适用场景，选择分库分表的实现方案。结合实际业务：学生（user表）定期参加体能测试（detect表），每一次体测之后，保留对应检测数据（data表），因此，数据data表中的核心数据：

分库分表第一步也是最重要的一步，即sharding column的选取，sharding column选择的好坏将直接决定整个分库分表方案最终是否成功。sharding column的选取跟业务强相关。

• 选择方法：分析你的API流量，将流量比较大的API对应的SQL提取出来，将这些SQL共同的条件作为sharding column。
• 选择示例：例如一般的OLTP系统都是对用户提供服务，这些API对应的SQL都有条件用户ID，那么，用户ID就是非常好的sharding column。

在上述学生体测业务中，我们需要汇总统计一次体测任务中，所有学生各项的体测结果，所以按照上述的原则，需要根据体测任务ID，即detect_id进行分表，以尽量减少在统计一次体测任务的数据时的跨表查询；但实际业务中，在学生端也有纵向对比的需求，即学生需要查看自己所有参加过的体测任务中的数据，这样的话，按照detect_id分表，再以user_id作为查询条件，就需要跨表查询，效率会很低。因此，最终方案是：不同字段冗余分表。

（1）冗余全量表

每个sharding列对应的表的数据都是全量的。以用户体测数据为例：分别使用三个独立的sharding column，即data_id（数据ID），detect_id（体测任务ID），user_id（学生ID）。

（2）冗余关系表选择

只有一个sharding column的分库分表的数据是全量的，其他分库分表只是与这个sharding column的关系表。实际使用中可能会冗余更多常用字段，如学生姓名、体测任务名称等。

（3）冗余全量表 VS 冗余关系表

• 速度对比：冗余全量表速度更快，冗余关系表需要二次查询，即使有引入缓存，还是多一次网络开销；
• 存储成本：冗余全量表需要几倍于冗余关系表的存储成本；
• 维护代价：冗余全量表维护代价更大，涉及到数据变更时，多张表都要进行修改。

选择冗余全量表还是索引关系表，这是一种架构上的权衡，两者的优缺点明显，在我们的业务中采用冗余全量表的方式。

非关系型数据库（ClickHouse）

上面提到的都是条件中有sharding column的SQL执行。但是，总有一些查询条件是不包含sharding column的，同时，我们也不可能为了这些请求量并不高的查询，无限制的冗余分库分表。另外，在分表前，我们会事先定义好分表的数量，随着业务扩张，单表数据达到大几千万甚至上亿，对于MySQL而言，还是不大友好的，再去增加分表数量，也是不大现实的。因此，专业的事情最好还是使用专业的工具-ClickHouse。

ClickHouse 是近年来备受关注的开源列式数据库，主要用于数据分析(OLAP)领域。目前国内社区火热，各个大厂纷纷跟进大规模使用:

• 今日头条内部用 ClickHouse 来做用户行为分析，内部一共几千个 ClickHouse 节点，单集群最大 1200 节点，总数据量几十 PB，日增原始数据 300TB 左右。
• 腾讯内部用 ClickHouse 做游戏数据分析，并且为之建立了一整套监控运维体系。
• 携程内部从 18 年 7 月份开始接入试用，目前 80% 的业务都跑在 ClickHouse 上。每天数据增量十多亿，近百万次查询请求。
• 快手内部也在使用 ClickHouse，存储总量大约 10PB， 每天新增 200TB， 90% 查询小于 3S。

在 1 亿数据集体量的情况下，ClickHouse 的平均响应速度是 Vertica 的 2.63 倍、InfiniDB 的 17 倍、MonetDB 的 27 倍、Hive 的 126 倍、MySQL 的 429 倍以及Greenplum 的 10 倍。

ClickHouse更多内容参考：https://juejin.cn/post/7120519057761107999

在 OLAP 数据库中，可变数据通常不受欢迎。ClickHouse 也不欢迎可变数据。然而现实情况，更新情况不可避免。比如，学生在体测过程中，是可以进行重复测试的，即需要进行更新数据。以下是关于clickhouse更新的解决方案：

参考：https://zhuanlan.zhihu.com/p/485645089

（1）Alter/Update Table

ClickHouse团队在2018年发布了UPDATE和DELETE，但是它不是原生的UPDATE和DELETE语句，而是被实现为ALTER TABLE UPDATE语句，如下所示：

ALTER TABLE [db.]table UPDATE column1 = expr1 [, ...] WHERE filter_expr;

如更新检测结果，ALTER UPDATE语句如下：

ALTER TABLE UPDATE detect_result=1 WHERE detect_id = 1 and user_id=4;

需要注意的是，ClickHouse的更新是一个异步的操作。当用户执行一个如上的Update操作获得返回时，ClickHouse内核其实只做了两件事情：

• 检查Update操作是否合法；
• 保存Update命令到存储文件中，唤醒一个异步处理merge和mutation的工作线程；

异步线程的工作流程极其复杂，总结其精髓描述如下：先查找到需要update的数据所在datapart，之后对整个datapart做扫描，更新需要变更的数据，然后再将数据重新落盘生成新的datapart，最后用新的datapart做替代并remove掉过期的datapart。

这就是ClickHouse对update指令的执行过程，可以看出，频繁的update指令对于ClickHouse来说将是灾难性的。(当然，我们可以通过设置，将这个异步的过程变成同步的过程，详细请看：Synchronicity of ALTER Queries，然而同步阻塞就会比较严重）。

（2）Incremental Log

Incremental log的思想是什么了？比如对于用户浏览统计表中的一条数据，如下所示：

┌──────────────UserID─┬─PageViews─┬─Duration─┬─Sign─┐ 
│ 4324182021466249494 │         5 │      146 │    1 │ 
└─────────────────────┴───────────┴──────────┴──────┘

现在有更新了：用户又浏览了一个页面，所以我们应该改变pageview从5到6，以及持续时间从146到185。那么按照Incremental log的思想，再插入两行:

┌──────────────UserID─┬─PageViews─┬─Duration─┬─Sign─┐ 
│ 4324182021466249494 │         5 │      146 │   -1 │ 
│ 4324182021466249494 │         6 │      185 │    1 │ 
└─────────────────────┴───────────┴──────────┴──────┘

第一个是删除行。它和我们已经得到的行是一样的只是Sign被设为-1。第二个更新行，所有数据设置为新值。之后我们有三行数据:

┌──────────────UserID─┬─PageViews─┬─Duration─┬─Sign─┐ 
│ 4324182021466249494 │         5 │      146 │    1 │ 
│ 4324182021466249494 │         5 │      146 │   -1 │ 
│ 4324182021466249494 │         6 │      185 │    1 │ 
└─────────────────────┴───────────┴──────────┴──────┘

那么对于count,sum,avg的计算方法如下：

-- number of sessions
count() -> sum(Sign)  
-- total number of pages all users checked 
sum(PageViews) -> sum(Sign * PageViews)  
-- average session duration, how long user usually spent on the website 
avg(Duration) -> sum(Sign * Duration) / sum(Sign)

这就是Incremental log方法，这种方法的不足之处在于：

• 首先需要获取到原数据，那么就需要先查一遍CK，或者将数据保存到其他存储中便于检索查询，然后我们才可以针对原数据插入一条 ‘delete’ rows；
• Sign operations在某些计算场景并不适合，比如min、max、quantile等其他场景；
• 额外的写入放大：当每个对象的平均更新次数为个位数时，更适合使用。

针对Incremental log方式的写入方案存储开销问题，clickhouse提供了CollapsingMergeTree，使用CollapsingMergeTree，“删除”行和旧的“删除”行将在合并过程中折叠。但是，注意这个引擎，只是解决了写放大问题，并不是说查询模式就不是Incremental Log这种，我们还是需要通过对sign的特殊计算方式，达到效果。

（3）Insert+xxxMergeTree

用Insert加特定引擎，也可以实现更新效果。该方法适用于xxxMergeTree，如ReplacingMergeTree或AggregatingMergeTree。但是了，更新是异步的。因此刚插入的数据，并不能马上看到最新的结果，因此并不是准实时的。

比如使用AggregatingMergeTree，用法如下：

CREATE TABLE IF NOT EXISTS whatever_table ON CLUSTER default (     
  user_id UInt64,
  gender SimpleAggregateFunction(anyLast, Nullable(Enum('女' = 0, '男' = 1))),
  ...
)
ENGINE = AggregatingMergeTree() partition by toYYYYMMDD(reg_date) ORDER BY user_id;

就以上建标语句展开分析，AggregatingMergeTree会将除主键（user）外的其余列，配合anyLast函数，替换每行数据为一种预聚合状态。其中anyLast聚合函数声明聚合策略为保留最后一次的更新数据。

实时性: 非准实时。

优点在于：
ClickHouse提供的这些mergeTree引擎，可以帮助我们达到最终一致性。
缺点在于：
xxxMergeTree并不能保证任何时候的查询都是聚合过后的结果，并且也没有提供标志位用于查询数据的聚合状态与进度。因此，为了确保数据在查询前处于已聚合的状态，还需手动下发optimize指令强制聚合过程的执行。

（4）Insert+xxxxMergeTree+Final

用xxxMergeTree是异步的，如何达到准实时的效果了？ClickHouse提供了FINAL关键字来解决这个问题。当指定FINAL后，ClickHouse会在返回结果之前完全合并数据，从而执行给定表引擎合并期间发生的所有数据转换。

用法

首先Insert数据：

INSERT INTO test_a (*) VALUES (1, 'a', 1) ;

查询时，加入final关键字，如下所示：

SELECT COUNT()FROM test_a FINAL

优缺点

对上述语句，explain后，查询执行计划如下所示：

Expression ((Projection + Before ORDER BY))
  Aggregating
    Expression (Before GROUP BY)
      SettingQuotaAndLimits (Set limits and quota after reading from storage)
        Expression (Remove unused columns after reading from storage)
          MergingFinal (Merge rows for FINAL)
            Expression (Calculate sorting key expression)
              ReadFromStorage (MergeTree with final)

从执行计划可以看出代价比较高：

• 是一个串行过程；
• 会进行分区合并；

因此，这个FINAL，也不宜频繁的使用。

总结

本文结合业务，寻求海量数据的解决方案。现有业务使用的是MySQL数据库，且数据量暂时可控，因此目前采用分库分表的策略。同时，也在为日益膨胀的数据做准备，拟采用ClickHouse，并使用Insert+ReplacingMergeTree及查询中去重的方案解决其更新问题。最后，欢迎有经验的伙伴多多指点！