项目背景

公司开发的一套用户广告归因系统，在用户点击广告后，需要记录用户的设备信息和广告的关系信息，方便后面做用户归因。广告的点击数据并发量巨大，每天的点击量在500W～1000W，点击数据我们通过 kafka 异步写入ClickHouse ，在写入ClickHouse数据时，常常报以下异常信息：

Error updating database. Cause: ru.yandex.clickhouse.except.ClickHouseException: ClickHouse exception, code: 252, host: 127.0.0.1, port: 8123; Code: 252, e.displayText() = DB::Exception: Too many parts (302). Merges are processing significantly slower than inserts (version 21.8.12.1)

原因：在数据插入到ClickHouse时，会生成parts文件，ClickHouse后台会有合并小文件的操作。当插入速度过快，生成parts小文件过快时，ClickHouse无法以适当的速度合并这些parts时会报上面这个错误。知道了问题的原因，就好解决问题了。我们最好是采用批量写入数据的方式而不要单条单条写入数据。

数据准备

click 表结构：

CREATE TABLE launch.click
(
    `id` UInt64,
    `app_type` UInt32,
    `product_type` UInt32,
    `channel_type` String,
    `agent_name` String,
    `advertiser_id` String,
    `aid` String,
    `aid_name` String,
    `cid` String,
    `cid_name` String,
    `ctype` String,
    `csite` String,
    `convert_id` String,
    `request_id` String,
    `sl` String,
    `imei` String,
    `idfa` String,
    `idfa_md5` String,
    `android_id` String,
    `oaid` String,
    `oaid_md5` String,
    `os` String,
    `mac` String,
    `mac1` String,
    `ip` String,
    `ipv6` String,
    `ua` String,
    `ts` String,
    `callback_param` String,
    `callback_url` String,
    `model` String,
    `caid` String,
    `caid_md5` String,
    `status` UInt32,
    `brand` String,
    `os_version` String,
    `version` UInt32,
    `data` String,
    `expire` DateTime,
    `create_time` DateTime COMMENT '创建时间'
)
ENGINE = ReplicatedReplacingMergeTree
PARTITION BY toYYYYMM(create_time)
PRIMARY KEY id
ORDER BY id

解决问题

发生问题的原因是我们的click数据是一条一条的写入到ClickHouse。每一次写入都会在底层生成 1 个 part 存储目录，后台任务会自动合并小 part 到一个大 part ，如果写入频次过高会出现 part 过多，merge 速度跟不上导致写入失败报错： Too many parts(301). Merges are processing significantly slower than inserts。所以我们要批量写入ClickHouse。Kafka批量消费消息，请参考： kafka实战一之批量消息消费。下面看下MyBatis的批量写入配置：

引入依赖Jar包

 			<!--mybatis-plus-->
        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>mybatis-plus-boot-starter</artifactId>
        </dependency>
        <!-- mybatis-plus 多数据源 -->
        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>dynamic-datasource-spring-boot-starter</artifactId>
        </dependency>

ClickMapper接口定义

@DS("launch")
public interface ClickMapper extends BaseMapper<ClickDO> {
    int batchInsert(@Param("list") List<ClickDO> list);
}

ClickMapper.xml 批量保存点击数据

    <insert id="batchInsert">
        insert into click(
        `id` ,
        `app_type` ,
        `product_type` ,
        `channel_type` ,
        `agent_name` ,
        `advertiser_id` ,
        `aid` ,
        `aid_name` ,
        `cid` ,
        `cid_name` ,
        `ctype` ,
        `csite` ,
        `convert_id` ,
        `request_id` ,
        `sl` ,
        `imei` ,
        `idfa` ,
        `idfa_md5` ,
        `android_id` ,
        `oaid` ,
        `oaid_md5` ,
        `os` ,
        `mac` ,
        `mac1` ,
        `ip` ,
        `ipv6` ,
        `ua` ,
        `ts` ,
        `callback_param` ,
        `callback_url` ,
        `model` ,
        `caid` ,
        `caid_md5` ,
        `status` ,
        `brand` ,
        `os_version` ,
        `version` ,
        `data` ,
        `expire` ,
        `create_time`
        )
        values
        <foreach collection="list" item="item" separator=",">
            (#{item.id},
            #{item.appType},
            #{item.productType},

            #{item.channelType},
            #{item.agentName},
            #{item.advertiserId},
            #{item.aid},
            #{item.aidName},
            #{item.cid},
            #{item.cidName},
             #{item.ctype},
             #{item.csite},

             #{item.convertId},
             #{item.requestId},

             #{item.sl},
             #{item.imei},
             #{item.idfa},
             #{item.idfaMd5},
             #{item.androidId},
             #{item.oaid},
             #{item.oaidMd5},

             #{item.os},
             #{item.mac},
             #{item.mac1},
             #{item.ip},
             #{item.ipv6},
             #{item.ua},
             #{item.ts},
             #{item.callbackParam},
             #{item.callbackUrl},
             #{item.model},
             #{item.caid},
             #{item.caidMd5},
             #{item.status},
             #{item.brand},
             #{item.osVersion},

             #{item.version},
             #{item.data},
             #{item.expire},
             #{item.createTime})
        </foreach>
    </insert>

ClickHouse写入规范

攒批写入：ClickHouse 必须攒批写入，至少 1000 条/批，建议 5k - 10w 一批写入ClickHouse，每一次写入都会在底层生成 1 个或者多个 part 存储目录，后台任务自动合并小 part 到一个大 part ，如果写入频次过高会出现 part 过多，merge 速度跟不上导致写入失败报错： Too many parts(301). Merges are processing significantly slower than inserts。

2. 减少分布式表直接写入：为了提高写入和查询性能，应尽可能直接写入本地表，而不是分布式表。写分布式表最终也会转发给本地表，但是分布式表存在写放大以及异步落盘消耗 IO 的问题，写入性能较差。

3. 约束数据一致性：ClickHouse 不支持数据写入的事务保证，因此需要通过外部导入数据模块来控制数据的幂等性。例如，如果某个批次的数据导入异常，可以删除对应的分区数据或清理导入的数据，然后重新导入该分区或批次的数据。也可以使用去重引擎（replacingMergetree）来保证最终一致性。

4. 大规模数据写入：如果需要进行大规模数据写入，建议提前拆分数据，并按节点均匀地写入 ClickHouse 的各个节点。如果存在特定的分布规则，可以在业务侧进行 hash 计算。

5. 一次只写入一个分区数据：为了避免写入性能下降和目录数量过多的问题，应该一次只写入一个分区的数据。如果一批写入数据跨多个分区，会导致底层产生多个 part 文件，消耗更多的 merge 性能，并且不利于幂等控制。

ClickHouse查询规范

高频过滤和点查询字段使用索引加速。

2. 避免使用select * 语句，应该明确需要查询的字段，只查询必要的字段。ClickHouse 底层是列式存储，查询的耗时与查询的字段大小和数量成线性关系。

3. 当查询千万以上的数据集时，建议使用where条件和limit语句来配合order by查询，以提高查询效率。

4. select {tablename} final 能够实现查询 ( read on merge )，但是会减慢查询速度，需要有针对性使用。

5. 尽量按分区过滤裁剪，通过指定分区字段可以减少底层数据库扫描的文件数量，提高查询性能。

6. 谨慎使用 delete 和 update 的 mutation 操作。ClickHouse 的 update 和 delete 是异步进行的，并且会重写 where 条件过滤出的数据 part ，是非常重的操作，可能会消耗较多系统资源。此外，update 和 delete 是按照 part 逐个执行，不会保证整体执行的原子性。

7. 如果对唯一性要求不高，可以采用近似去重 uniqCombined 来优化去重逻辑，从而提高十倍的查询性能。如果查询允许有误差，可以使用 uniqCombined 替代，否则应该继续使用 distinct 语法。使用 distinct 会对查询性能有一定影响。

建表规范

1. 高可用集群不可创建非 Replicated 表，非高可用集群不可创建 Replicated 表。

2. 如果对数据最终一致性有强要求，需要使用 ReplacingMergeTree 或者 CollapsingMergeTree 引擎，并定期进行 optimize 或使用 select {tablename} final 实现最终去重。

3. 在规划分区时，应该合理规划分区个数，并尽可能利用分区。一张表分区数不建议超过 1000 个，可以在查询时有效帮助进行数据过滤，使用得当可以提升数倍查询性能，通常按天分区是比较普遍的做法。分区也不建议过多，因为 ClickHouse 不同分区的数据不会合并，容易导致 part 过多，从而导致查询和重启变得很慢。

4. 建表时尽可能提前规划好表字段，并尽量避免删改字段。删改字段会重写整个表的全量数据，对于大表会消耗大量资源，执行时间可能很长。此外，删改字段期间也容易阻塞其他 DDL 语句，影响表的 merge 操作。如果中途出错，有概率会导致不可预知的数据一致性问题。

5. 禁止修改索引列，对索引列的修改会导致现有索引失效，触发重建索引，期间查询数据不准确。

6. 约束 COS 上存储数据的量，尽可能避免对冷分区进行写入和 mutation 操作。COS 单个桶大约只有1GB的带宽，远低于多节点的本地盘和云盘性能，且网络延迟比较高。如果 COS 上存储过多数据，会严重影响查询效率。针对 COS 分区的写入时，会触发 COS 分区进行 merge ，merge 效率也会降低甚至会影响本地盘的数据操作。