1、背景

魔笛活动平台目前在采集每个活动的用户行为数据并进行查询，解决线上问题定位慢，响应不及时的问题，提升客诉的解决效率。目前每天采集的数据量5000万+，一个月的数据总量15亿+，总数据量40亿+，随着接入的活动越来越多，采集上报的数据量也会越来越大。目前采用ClickHouse来存储数据，可以在秒级别内处理数十亿条数据，能够达到50MB-200MB/s的写入吞吐能力，按照每行100Byte估算，大约相当于50W-200W条/s的写入速度。这里关于ClickHouse就不再赘述，感兴趣的可以看上篇文章。这里是我在实际使用过程中，发现了一些写入和查询的相关问题，并进行了相应的优化。

2、写入优化

2.1 历史方案

为了更好的收集活动数据，自己开发了一款埋点sdk用于收集各业务的活动埋点数据，埋点数据统一异步发送MQ，然后活动平台消费MQ数据，经过一定处理后通过MybatisPlus方式批量写入clickhouse，每次批量写入5000条。

2.2 出现的问题

当消费MQ的TPS超过3000的时候，出现了以下问题：

1. 出现报错：写入量大的时候会报这个错误：too many parts。

2. 单次写入条数少：我明明配置的每次批量写入5000条，但是每次基本都是一条一条写入。

3. 性能很差：单次写入最大耗时居然有250000毫秒，平均也超过50000毫秒！

2.3 出现问题的原因

（1）单次写入条数少原因：MybatisPlus的savebatch单次最大写入SQL是4M，按照单条活动的数据大小，最大单次也就写1000条数据，再多就会一次一次写入。所以虽然配置的是每次写入5000，但是实际看是每次写入一条或者几条。

（2）too many parts错误：clickhouse操作数据的最小操作单元是block，每次写入都会按照zookeeper记录的唯一自增的blockId，按照PartitionId_blockId_blockId_0生成data parts，也就是小文件，然后后台会有merge线程，不定时（分钟级别）的将多个小文件进行合并，生成PartitionId_MinBlockNum_MaxBlockBum_Level的文件，未达到data parts最小rows或者大小限制前，会持续merge，每次merge的耗时大概5分钟左右。由于merge线程池是固定的，默认32，所以如果插入过于频繁，merge压力过大，处理不了，就会出现too many parts的报错。例如并发数为 200，这样一批写入到 ClickHouse 中就会产生 200 个文件，几批下来如果 ClickHouse 内部线程没来及合并相同分区，就会抛异常。而 ClickHouse 默认一次合并超过 300 个文件就会报错。

（3）性能差的原因：因为写clickhouse底层都是使用httpclient的方式写入的，所以对于clickhouse来说单条频繁写入效率很低。

2.4 改造方案

（1）写入clickhouse的并发数调小，批处理的数据size间隔调大，比如之前200并发调整到50并发，从之前一批1条数据调整到10000条数据，clickhouse官网建议每批次写入100000+条（要视flink TM 内存大小调整，防止批量过大出现oom）。从而减少clickhouse文件的个数，避免超过parts_to_throw_insert默认值。一般最好一秒钟写入一次clickhouse。

（2）将由MybatisPlus的savebatch批量写入改为其他方式写入。采用clickhouse原生的jdbc写入或者flink摄入，flink我这边自定义了sink 用于摄入clickhouse，达到一定批次或者执行checkpoint时就写入一次。

（3）为了保证批量，我这边采用实时双buffer缓冲队列方式写入，这个队列可以是本地缓存队列，也可以是redis缓存队列。根据时间窗口期和固定写入阈值（针对波动大的可以按照二次指数平滑函数去确定阈值）进行写入与否的判断。设置一个读队列，一个写队列，并设置一个开关，一个阈值，一个定时器，当数据来时，默认放入写队列中，当队列中数据的数量大于阈值，将开关关闭，将写队列数据放到读队列中，从读队列拿出数据批量写入clickhouse，将开关打开，清空队列中数据。如果队列中数据在一定时间内，比如10秒，一直没有达到阈值，也关闭开关，写队列数据放到读队列中，从读队列拿出数据批量写入clickhouse，将开关打开，清空队列中数据。