Shopee|Shopee ClickHouse 冷热数据分离存储架构与实践 ShopeeClickHouse冷热数据分离存储

本文首发于微信公众号“ Shopee技术团队”。

摘要

Shopee ClickHouse 是一款基于开源数据库 ClickHouse 做二次开发、架构演进的高可用分布式分析型数据库。本文将主要介绍 Shopee ClickHouse 的冷热分离存储架构和支持公司业务的实践。
Shopee ClickHouse 的冷热分离存储架构使用 JuiceFS 客户端 mount 远端对象存储到本地机器路径，通过编写 ClickHouse 的存储策略，如同使用多卷存储一样使用远端对象存储。因为我们用同一个 ClickHouse DB 集群支持多个团队的业务，不同团队甚至相同团队的不同业务之间对数据的冷热划分基准可能都不同，所以在做冷热分离时策略需要做到 ClickHouse 的表级别。
为了做到表级别的冷热分离，我们依照提前编辑好的存储策略，针对存量需要做冷热隔离的业务表，修改表的存储策略。对于新的需要做冷热分离的业务表，建表时指明使用支持数据落在远端存储的存储策略，再通过细化 TTL 表达式判断数据应该落在本地还是远端。
冷热分离存储架构上线后，我们遇到了一些问题和挑战，比如：juicefs object request error、Redis 内存增长异常、suspicious broken parts 等。本文会针对其中一些问题，结合场景上下文，并通过源码分析来给出解决方案。
总的来说 Shopee ClickHouse 冷热存储架构的整体设计思想是：本地 SSD 存储查询热数据，远端存储查询相对不那么频繁的数据，从而节约存储成本，支持更多的数据存储需求。

1. Shopee ClickHouse 集群总架构 ClickHouse 是一款开源的列存 OLAP（在线分析查询）型数据库，实现了向量化执行引擎，具有优秀的 AP 查询性能。Shopee ClickHouse 则是基于 ClickHouse 持续做二次迭代开发和产品架构演进的分析型数据库。
下图展示了 Shopee ClickHouse DB 集群的架构：

文章图片

从上到下依次是用户请求介入 SLB、Proxy 层、ClickHouse DB 集群层，最下方是远端对象存储，这里我们用的是 Shopee STO 团队提供的 S3。
其中，SLB 提供用户请求路由；Proxy 层提供了查询路由，请求会根据用户连接串中的集群名，路由到对应的集群中，也提供了部分写入 balance 和查询路由的能力；ClickHouse DB 集群层是由 Shopee ClickHouse 数据库组成的分布式集群，目前有以 SSD 磁盘作为热数据存储介质的计算型分布式集群，和计算型单节点集群，还有以 SATA Disk 作为存储介质的存储型分布式集群；最下方的远端存储则用作冷数据存储介质。
2. 冷热分离存储架构方案用户希望数据可以存储得更多更久，查询速度更快。但是通常数据存储得越多，在相同查询条件下，返回延时就会越高。
从资源利用率上来说，我们希望存储在 Shopee ClickHouse 上的数据可以被更多地访问和利用，为业务提供更广泛的支持。所以，起初我们要求业务方存储到 Shopee ClickHouse 数据库中的数据是用户的业务热数据。
但是这样也带来了一些问题，比如：用户有时候需要查询时间相对久一点的数据做分析，这样就得把那部分不在 ClickHouse 的数据导入后再做分析，分析结束后还要删除这部分数据。再比如：一些通过日志服务做聚合分析和检索分析的业务，也需要相对久一点的日志服务数据来帮助监管和分析日常业务。
基于此类需求，我们一方面希望资源的最大化利用，一方面希望支持更多的数据存储量，同时不影响用户热数据的查询速度，所以使用冷热数据分离的存储架构就是一个很好的选择。
通常，冷热分离方案的设计需要考虑以下几个问题：

如何存储冷数据？
如何高效稳定简单地使用冷存介质？
热数据如何下沉到冷存介质？
架构的演进如何不影响现有的用户业务？

而冷数据存储介质的选择一般通过以下几个要点做对比分析：

成本
稳定性
功能齐全（数据在下沉过程中依然可以被正确查询，数据库的数据也可以被正确写入）
性能
扩展性

2.1 冷存介质的选择和 JuiceFS 可以用作冷存储的介质一般有 S3、Ozone、HDFS、SATA Disk。其中，SATA Disk 受限于机器硬件，不易扩展，可以先淘汰。而 HDFS、Ozone 和 S3 都是比较好的冷存介质。
同时，为了高效简单地使用冷存介质，我们把目光锁定在了 JuiceFS 上。JuiceFS 是一种基于 Redis 和云对象存储构建的开源 POSIX 文件系统，可以使我们更加便捷和高效地访问远端对象存储。
JuiceFS 使用公有云中已有的对象存储，如 S3、GCS、OSS 等。用 JuiceFS 做存储，数据实际上存储在远端，而 JuiceFS 重点关注这些存储在远端的数据文件的元数据管理。JuiceFS 选择 Redis 作为存储元数据的引擎，这是因为 Redis 存储都在内存中，可以满足元数据读写的低延时和高 IOPS，支持乐观事务，满足文件系统对元数据操作的原子性[1]。
JuiceFS 提供了一种高效便捷的远端存储访问方式，只需要通过 JuiceFS 的客户端，使用 format 和 mount 命令，就可以将远端存储 mount 到本地路径。我们 ClickHouse 数据库访问远端存储就可以如同访问本地路径一样访问。
选择了 JuiceFS 后，我们再把目光转回冷数据存储介质的筛选。由于 JuiceFS 主要支持的后台存储层为对象存储类别，余下的选项变成了 S3 和 Ozone。我们设计了一个如下的 benchmark , 使用 ClickHouse TPCH Star Schema Benchmark 1000s（benchmark 详细信息可以参照 ClickHouse 社区文档[2]）作为测试数据，分别测试 S3 和 Ozone 的 Insert 性能，并使用 Star Schema Benchmark 的 select 语句做查询性能对比。
查询的数据处于以下三种存储状态：

一部分在 Ozone/S3，一部分在本机 SSD 磁盘；
全部在 Ozone/S3；
全部在 SSD 上。

以下是我们的测试抽样结果：
（1）Insert 性能抽样结果
Insert Lineorder 表数据到 Ozone:

文章图片

Insert Lineorder 表数据到 S3:

文章图片

可以看出，S3 的 Insert 性能稍微强势一点。
（2）查询性能抽样结果
根据 ClickHouse Star Schema Benchmark，在导入完毕 Customer、Lineorder、Part、Supplier 表后，需要根据四张表的数据创建一个打平的宽表。

CREATE TABLE lineorder_flat ENGINE = MergeTree PARTITION BY toYear(LO_ORDERDATE) ORDER BY (LO_ORDERDATE, LO_ORDERKEY) AS SELECT l.LO_ORDERKEY AS LO_ORDERKEY, l.LO_LINENUMBER AS LO_LINENUMBER, l.LO_CUSTKEY AS LO_CUSTKEY, l.LO_PARTKEY AS LO_PARTKEY, l.LO_SUPPKEY AS LO_SUPPKEY, l.LO_ORDERDATE AS LO_ORDERDATE, l.LO_ORDERPRIORITY AS LO_ORDERPRIORITY, l.LO_SHIPPRIORITY AS LO_SHIPPRIORITY, l.LO_QUANTITY AS LO_QUANTITY, l.LO_EXTENDEDPRICE AS LO_EXTENDEDPRICE, l.LO_ORDTOTALPRICE AS LO_ORDTOTALPRICE, l.LO_DISCOUNT AS LO_DISCOUNT, l.LO_REVENUE AS LO_REVENUE, l.LO_SUPPLYCOST AS LO_SUPPLYCOST, l.LO_TAX AS LO_TAX, l.LO_COMMITDATE AS LO_COMMITDATE, l.LO_SHIPMODE AS LO_SHIPMODE, c.C_NAME AS C_NAME, c.C_ADDRESS AS C_ADDRESS, c.C_CITY AS C_CITY, c.C_NATION AS C_NATION, c.C_REGION AS C_REGION, c.C_PHONE AS C_PHONE, c.C_MKTSEGMENT AS C_MKTSEGMENT, s.S_NAME AS S_NAME, s.S_ADDRESS AS S_ADDRESS, s.S_CITY AS S_CITY, s.S_NATION AS S_NATION, s.S_REGION AS S_REGION, s.S_PHONE AS S_PHONE, p.P_NAME AS P_NAME, p.P_MFGR AS P_MFGR, p.P_CATEGORY AS P_CATEGORY, p.P_BRAND AS P_BRAND, p.P_COLOR AS P_COLOR, p.P_TYPE AS P_TYPE, p.P_SIZE AS P_SIZE, p.P_CONTAINER AS P_CONTAINER FROM lineorder AS l INNER JOIN customer AS c ON c.C_CUSTKEY = l.LO_CUSTKEY INNER JOIN supplier AS s ON s.S_SUPPKEY = l.LO_SUPPKEY INNER JOIN part AS p ON p.P_PARTKEY = l.LO_PARTKEY

再执行这条 SQL 语句，当数据全部在 Ozone 上时，发生了如下 Error：

Code: 246. DB::Exception: Received from localhost:9000. DB::Exception: Bad size of marks file '/mnt/jfs/data/tpch1000s_juice/customer/all_19_24_1/C_CUSTKEY.mrk2': 0, must be: 18480

Select 数据一部分在 Ozone，并且此过程中发生了数据从 SSD 磁盘下沉到 Ozone 的情况。
结果：Hang 住，无法查询。

做这个测试时，我们使用的 Ozone 是社区版本 1.1.0-SNAPSHOT，此次测试结果仅说明 Ozone 1.1.0-SNAPSHOT 不是很适合我们的使用场景。

由于 Ozone 1.1.0-SNAPSHOT 在我们的使用场景中有功能性的缺点，所以后续的 Star Schema Benchmark 的性能测试报告重点放在 SSD 和 S3 的性能对比上（详细 Query SQL 语句可以从 ClickHouse 社区文档获取）。

Query No.	Query Latency Data on JuiceFS	Query Latency Data on ? JuiceFs + ? SSD	Query Latency Data on SSD
Q1.1	8.884 s	8.966 s	1.417 s
Q1.2	0.921 s	0.998 s	0.313 s
Q1.3	0.551 s	0.611 s	0.125 s
Q2.1	68.148 s	36.273 s	5.450 s
Q2.2	54.360 s	20.846 s	4.557 s
Q2.3	55.329 s	22.152 s	4.297 s
Q3.1	60.796 s	27.585 s	7.999 s
Q3.2	67.559 s	29.123 s	5.928 s
Q3.3	45.917 s	20.682 s	5.606 s
Q3.4	0.675 s	0.202 s	0.188 s
Q4.1	100.644 s	41.498 s	7.019 s
Q4.2	32.294 s	2.952 s	2.464 s
Q4.3	33.667 s	2.813 s	2.357 s

最终，在各个方面的对比下，我们选择 S3 作为冷存介质。
因此，冷热存储分离的方案采用 JuiceFS+S3 实现，下文将简述实现过程。
2.2 冷热数据存储分离的实现首先，我们通过使用 JuiceFS 客户端，mount S3 bucket 到本地存储路径 /mnt/jfs，然后编辑 ClickHouse 存储策略配置 ../config.d/storage.xml 文件。编写存储策略配置文件时要注意，不要影响到历史用户存储（即保留之前的存储策略）。在这里，default 就是我们的历史存储策略，hcs_ck 是冷热分离的存储策略。
详细信息可以参照下图：

文章图片

有需要冷热分离存储的业务，只需要在建表 Statement 里面写明存储策略为 hcs_ck，然后通过 TTL 的表达式来控制冷数据下沉策略。
下面通过一个例子说明使用方式和数据分离过程。表 hcs_table_name 是一个需要冷热存储分离的业务日志数据表，以下是建表语句：

CREATE TABLE db_name.hcs_table_name ( ..... `log_time` DateTime64(3), `log_level` String, ..... `create_time` DateTime DEFAULT now() ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{layer}-{shard}/db_name.hcs_table_name ', '{replica}') PARTITION BY toYYYYMMDD(log_time) ORDER BY (ugi, ip) TTL toDateTime(log_time) TO VOLUME 'v_ssd', toDateTime(log_time) + toIntervalDay(7) TO VOLUME 'v_cold', toDateTime(log_time) + toIntervalDay(14) SETTINGS index_granularity = 16384, storage_policy = 'hcs_ck', parts_to_throw_insert = 1600

通过 TTL 表达式可以看到，hcs_table_name 这个表指明最近 7 天的数据存储在本地 SSD 磁盘，第 8 到 14 天的数据存储在远端 S3，超过 14 天的数据过期删除。
大体流程如下图所示：

文章图片

表 hcs_table_name 的 data parts（ClickHouse 的数据存储以 data part 为基本处理单位）会被后台任务调度，后台任务由线程 BgMoveProcPool 执行，这个线程来自 back_ground_move_pool（注意和 back_ground_pool 不是同一个）。

std::optional background_move_pool; /// The thread pool for the background moves performed by the tables.

后台任务调度会判断 data parts 是否需要 move（数据是否需要下沉移动到远端存储上）和是否可以 move。
如果需要执行 move，后台 move_pool 会创建一个 move 的 task。这个 task 的核心逻辑是：首先选择需要 move 的 data parts，然后再 move 这些 data parts 到目的存储。
在接口：

MergeTreePartsMover::selectPartsForMove

中根据 TTL Expression 表达式获取 ttl_entry，然后根据 data parts 中的 ttl_move 信息，选出需要 move 的 data parts，存储 data parts 的 move_entry（包含 IMergeTreeDataPart 指针和需要预留的存储空间大小）到 vector 中。之后会调用接口：

MergeTreeData::moveParts

实现 move 操作，move 的过程简单来说就是 clone SSD 磁盘上的 data parts 到远端存储 S3 上 hcs_table_name 表的 detach 目录下，然后再从 detach 目录下把 data parts 移出来，最后这些在 SSD 磁盘上的 data parts 会在 IMergeTreeDataPart 的析构函数中被清除。
所以整个 move 过程中，表一直是可查的，因为是 clone 操作，同一时刻下 move 的 data parts 要么在 SSD 磁盘上为 active，要么在远端存储上为 active。
关于表 data parts 的 move 信息，也可以查询系统表 system.parts 的以下三个字段：

move_ttl_info.expression； move_ttl_info.min; move_ttl_info.max;

3. 实践分享在 Shopee ClickHouse 冷热数据分离存储架构上线后，我们总结了一些实践中遇到的问题。
3.1 Redis 内存增长异常 S3 上的数据存储量并没有增加太多，Redis 内存却持续高速增长。
JuiceFS 使用 Redis 存储 S3 上的数据文件的元数据，所以正常情况下，S3 上的数据文件越多，Redis 存储使用量也就越多。一般这种异常情况是因为目标表有很多小文件没有 merge 而直接下沉，很容易打满 Redis。
这也会引入另一个问题：一旦 Redis 内存打满，JuiceFS 就不能再成功写数据到 S3 上，如果 unmount 掉 JuiceFS 客户端，也无法再次成功 mount 上去，再次 mount 的时候会抛 Error：

Meta: create session: OOM command not allowed when used memory > 'maxmemory'.

要避免这种问题发生，首先应该做好 ClickHouse merge 状态的监控。clickhouse-exporter 会采集一个 merge 指标 clickhouse_merge，这个指标会采集到当前正在触发的 merge 个数（通过查询 system.metrics 表 metric=‘merge’），每触发一次 merge 会有一个表的多个 data parts 做合并操作。按照我们的经验来看，若每三个小时 merge 的平均次数小于 0.5，那么很有可能是这台机器的 merge 出现了问题。
而 merge 异常的原因可能有很多（例如 HTTPHandler 线程、ZooKeeperRecv 线程持续占据了大量 CPU 资源等），这个不是本文的介绍重点，在此不再展开。所以可以设置告警规则，如果三小时内 merge 次数小于 0.5 次，告警给 ClickHouse 的开发运维团队同学，避免大量小文件产生。
如果已经有大量小文件下沉到 S3 应该怎么办？
首先要阻止数据继续下沉，可以通过两种方式找到有大量小文件下沉的用户业务表。
第一种方式：查看 ClickHouse 的 Error Log，找到抛 too many parts 的表，再进一步判断抛 Error 的表是否有冷热存储。
第二种方式：通过查询 system.parts 表，找出 active parts 明显过多，并且 disk_name 等于冷存的别名的。定位到产生大量小文件的表后，通过 ClickHouse 系统命令 SQL：

SYSTEM STOP MOVES [[db.]merge_tree_family_table_name]

停止数据继续下沉，避免 Redis 内存打满。
如果表比较小，比如压缩后小于 1TB（这里的 1TB 是一个经验值，我们曾经使用 insert into ... select * from … 方式导表数据，如果大于 1TB，导入时间会很久，还有一定的可能性在导入中途失败），在确认 merge 功能恢复正常后，可以选择创建 temp table > insert into this temp table > select * from org table，然后 drop org table > rename temp table to org table。
如果表比较大，确认 merge 功能恢复正常后，尝试通过系统命令 SQL：

SYSTEM START MERGES [[db.]merge_tree_family_table_name]

唤醒 merge 线程。如果 merge 进行缓慢，可以查询 system.parts 表，找到已经落在 S3 上的 data parts，然后手动执行 Query：

ALTER TABLE table_source MOVE PART/PARTITION partition_expr TO volume 'ssd_volume'

将落在 S3 上的小文件移回到 SSD 上。因为 SSD 的 IOPS 比 S3 要高很多（即使是通过 JuiceFS 访问加速后），这样一方面加快 merge 过程，一方面因为文件移出 S3，会释放 Redis 内存。
3.2 JuiceFS 读写 S3 失败数据下沉失败，通过 JuiceFS 访问 S3，无法对 S3 进行读写操作，这个时候用户查询如果覆盖到数据在 S3 上的，那么查询会抛 S3 mount 的本地路径上的数据文件无法访问的错误。遇到这个问题可以查询 JuiceFS 的日志。

JuiceFS 的日志在 Linux CentOS 中存储在 ??syslog 上，查询日志可以用方法 cat/var/log/messages|grep 'juicefs'，不同操作系统对应的日志目录可以参照 JuiceFS 社区文档[3]。

我们遇到的问题是 send request to S3 host name certificate expired。后来通过联系 S3 的开发运维团队，解决了访问问题。
那么如何监控这类 JuiceFS 读写 S3 失败的情况呢？可以通过 JuiceFS 提供的指标 juicefs_object_request_errors 监控，如果出现 Error 就告警团队成员，及时查询日志定位问题。
3.3 clickhouse-server 启动失败对历史表需要做冷热数据存储分离的复制表（表引擎含有 Replicated 前缀）修改 TTL 时，clickhouse-server 本地 .sql 文件元数据中的 TTL 表达式和 ZooKeeper 上存储的 TTL 表达式不一致。这个是我们在测试过程中遇到的问题，如果没有解决这个问题而重启 clickhouse-server 的话，会因为表结构没有对齐而使 clickhouse-server 启动失败。
这是因为对复制表的 TTL 的修改是先修改 ZooKeeper 内的 TTL，然后才会修改同一个节点下的机器上表的 TTL。所以如果在修改 TTL 后，本地机器 TTL 还没有修改成功，而重启了 clickhouse-server，就会发生上述问题。
3.4 suspicious_broken_parts 重启 clickhouse-server 失败，抛出 Error：

DB::Exception: Suspiciously many broken parts to remove

这是因为 ClickHouse 在重启服务的时候，会重新加载 MergeTree 表引擎数据，主要代码接口为：

MergeTreeData::loadDataParts(bool skip_sanity_checks)

在这个接口中会获取到每一个表的 data parts，判断 data part 文件夹下是否有 #DELETE_ON_DESTROY_MARKER_PATH 也就是 delete-on-destroy.txt 文件存在。如果有，将该 part 加入到 broken_parts_to_detach，并将 suspicious_broken_parts 统计个数加 1。
那么在冷热数据存储分离的场景下，data parts 通过 TTL 做下沉的时候，在核心接口 move 操作的函数中会有如下的代码调用关系：

MergeTreeData::moveParts->MergeTreePartsMover::swapClonedPart->MergeTreeData::swapActivePart

在最后一个函数中交换 active parts 的路径指向，也就是上文说的，data parts 在 move 过程中，数据是可查的，要么在 SSD 为 active，要么在 S3 为 active。

void MergeTreeData::swapActivePart(MergeTreeData::DataPartPtr part_copy) { auto lock = lockParts(); for (auto original_active_part : getDataPartsStateRange(DataPartState::Committed)) // NOLINT (copy is intended) { if (part_copy->name == original_active_part->name) { ..... String marker_path = original_active_part->getFullRelativePath() + DELETE_ON_DESTROY_MARKER_PATH; try { disk->createFile(marker_path); } catch (Poco::Exception & e) ... }

在这个接口中，旧的 active parts（也就是 replacing parts）内会创建 #DELETE_ON_DESTROY_MARKER_PATH 文件来把 state 修改为 DeleteOnDestory，用于后期 IMergeTreeDataPart 析构时删除该 state 的 data parts。
这也就是在我们的使用场景下会出现 suspicious_broken_parts 的原因，这个值超过默认阈值 10 的时候就会影响 ClickHouse 服务启动。
解决方案有两种：第一种，删除这个机器上抛出该错误的表的元数据 .sql 文件、存储数据、ZooKeeper 上的元数据，重启机器后重新建表，数据会从备份机器上同步过来。第二种，在 ClickHouse /flags 路径下用 clickhouse-server 进程的运行用户创建 force_restore_data flag，然后重启即可。
从上述问题中可以看到，使用 JuiceFS+S3 实现了冷热数据分离存储架构后，引入了新的组件（JuiceFS+Redis+S3），数据库的使用场景更加灵活，相应地，各个方面的监控信息也要做好。这里分享几个比较重要的监控指标：

JuiceFS：juicefs_object_request_errors：JuiceFS 对 S3 读写的健康状态监控。
Redis：Memory Usage：监控 Redis 的内存使用情况。
ClickHouse：clickhouse_merge：监控集群中机器的 merge 状态是否正常。

4. 冷热存储架构收益总述冷热数据存储分离后，我们更好地支持了用户的数据业务，提高了整体集群的数据存储能力，缓解了各个机器的本地存储压力，对业务数据的管理也更加灵活。
冷热数据分离架构上线前，我们的集群机器平均磁盘使用率接近 85%。上线后，通过修改业务用户表 TTL，这一数据下降到了 75%。并且整体集群在原有的业务量基础上，又支持了两个新的数据业务。如果没有上线冷热隔离，我们的集群在扩容前就会因为磁盘用量不足而无法承接新的项目。当前我们下沉到远端 S3 的数据量大于 90TB（压缩后）。
未来 Shopee ClickHouse 会持续开发更多有用的 feature，也会持续演进产品架构。目前 JuiceFS 在我们生产环境中的使用非常稳定，我们后续会进一步使用 JuiceFS 访问 HDFS，进而实现 Shopee ClickHouse 存储计算分离架构。

本文提到的各个产品组件版本信息如下：

Shopee ClickHouse：当前基于社区版 ClickHouse 20.8.12.2-LTS version

JuiceFS：v0.14.2

Redis：v6.2.2，sentinel model，开启 AOF（策略为 Every Secs），开启 RDB（策略为一天一备份）

S3：由 Shopee STO 团队提供

Ozone：1.1.0-SNAPSHOT