clickhouse 在order by非常慢的情况下优化（引擎+分段sql）

Posted 2021-09-20 寒江孤影，江湖过客，相逢何必曾相识

 

 

1、展示一下order by 在上亿级别数据量有多慢

对于clickhouse来说，当表的基础大到亿级别, 如果做查询后，在做order by 速递是非常慢的;

比如我有一张表有3亿条数据，表结构是：

{

　　dvid 设备ID ，
　　json 每个设备的信息 ,

　　dt 时间 ,

　　filter 过滤条件 ，

　　order_key 排序条件

}

ENGINE = MergeTree
PARTITION BY toYYYYMMDD(toDate(dt))
PRIMARY KEY (dt, dvid)
ORDER BY (dt, dvid)
SETTINGS index_granularity = 256

现在要统计3个月（1.5亿条）的设备数量，并按照order_key做排序

给出的sql：

select count(1) from 
(select 
dt,
toUInt32(dvid) as dvid ,
json 
from
表
where filter条件
order by order_key排序条件)

很简单的sql，但是查询速度很慢：

+---------+
| count() |
+---------+
| 1262100 |
+---------+
1 row in set (37.50 sec)

但是当你把order by去掉后，同样是数据量查询速度会非常快：

+---------+
| count() |
+---------+
| 1262100 |
+---------+
1 row in set (1.74 sec)

 

2、查看执行计划，看看为什么这么慢

2.1、先看没有order by的执行计划：

+-------------------------------------------------------------------------------------------------+
| explain                                                                                         |
+-------------------------------------------------------------------------------------------------+
| Expression ((Projection + Before ORDER BY))                                                     |
|   Aggregating                                                                                   |
|     Expression ((Before GROUP BY + Projection))                                                 |
|       SettingQuotaAndLimits (Set limits and quota after reading from storage)                   |
|         Union                                                                                   |
|           Expression ((Convert block structure for query from local replica + Before ORDER BY)) |
|             SettingQuotaAndLimits (Set limits and quota after reading from storage)             |
|               ReadFromStorage (MergeTree)                                                       |
|           ReadFromPreparedSource (Read from remote replica)                                     |
+-------------------------------------------------------------------------------------------------+
9 rows in set (0.05 sec)

大概意思是：

第一步：从每个机器的硬盘上把数据读取出来

第二步：union一下每个机器读取的数据

第三步：求count

没有order by的总耗时：1.7s

2.2、进行order by的执行计划

+-------------------------------------------------------------------------------------------+
| explain                                                                                   |
+-------------------------------------------------------------------------------------------+
| Expression ((Projection + Before ORDER BY))                                               |
|   Aggregating                                                                             |
|     Expression ((Before GROUP BY + Projection))                                           |
|       MergingSorted (Merge sorted streams for ORDER BY)                                   |
|         SettingQuotaAndLimits (Set limits and quota after reading from storage)           |
|           Union                                                                           |
|             Expression (Convert block structure for query from local replica)             |
|               FinishSorting                                                               |
|                 Expression (Before ORDER BY)                                              |
|                   SettingQuotaAndLimits (Set limits and quota after reading from storage) |
|                     Expression (Remove unused columns after reading from storage)         |
|                       Union                                                               |
|                         MergingSorted (Merge sorting mark ranges)                         |
|                           Expression (Calculate sorting key prefix)                       |
|                             ReadFromStorage (MergeTree with order)                        |
|                         MergingSorted (Merge sorting mark ranges)                         |
|                           Expression (Calculate sorting key prefix)                       |
|                             ReadFromStorage (MergeTree with order)                        |
|                         MergingSorted (Merge sorting mark ranges)                         |
|                           Expression (Calculate sorting key prefix)                       |
|                             ReadFromStorage (MergeTree with order) 
                       ...................
　　　　　　　　　　　　　　　　省略一摸一样的N行

　　　　　　　　　　　　　　...................

大概意思：

1、从硬盘读取需要的数据（部分，因为order by需要在内存里面快速排序，无法读取全部）

2、按照order by 的key进行排序

3、N多个order by排序完的数据，在做最终汇总，然后对汇总后的数据在做排序（这一步也会根据数据量分成多步完成）

4、最终做聚合求count

最终order by的总耗时：37.5s

3、优化

3.1、切换引擎

上一步慢、猜测可能是clickhouse的mergeTree引擎并不适合做排序操作,于是尝试了各种引擎,最终得到最适合order by的引擎：ReplicatedAggregatingMergeTree

重建建表：

ENGINE = ReplicatedAggregatingMergeTree(\'/clickhouse/tables/{shard}/库/表\', \'{replica}\')
PARTITION BY toYYYYMMDD(toDate(dt))
PRIMARY KEY dvid
ORDER BY dvid
SETTINGS index_granularity = 8196

然后增加表的数据量，将原来的1.5亿条 增加到3.5亿条

执行同样的语句，测试性能：

1、不带order by

+---------+
| count() |
+---------+
| 3290193 |
+---------+
1 row in set (3.58 sec)

2、带order by的查询速度

+---------+
| count() |
+---------+
| 3290193 |
+---------+
1 row in set (4.16 sec)

 

3.2、查看切换ReplicatedAggregatingMergeTree引擎后的执行计划

1、不带order by的执行计划

+-------------------------------------------------------------------------------------------------+
| explain                                                                                         |
+-------------------------------------------------------------------------------------------------+
| Expression ((Projection + Before ORDER BY))                                                     |
|   Aggregating                                                                                   |
|     Expression ((Before GROUP BY + Projection))                                                 |
|       SettingQuotaAndLimits (Set limits and quota after reading from storage)                   |
|         Union                                                                                   |
|           Expression ((Convert block structure for query from local replica + Before ORDER BY)) |
|             SettingQuotaAndLimits (Set limits and quota after reading from storage)             |
|               ReadFromStorage (MergeTree)                                                       |
|           ReadFromPreparedSource (Read from remote replica)                                     |
+-------------------------------------------------------------------------------------------------+
9 rows in set (0.05 sec)

不带order by的执行计划MergeTree的引擎一样，所以没任何改变，之所以查询速度从1.74s变到3.58s 是因为基础数据从原来的1.5亿变成现在的3.3亿

2、带order by的执行计划

+-----------------------------------------------------------------------------------------------+
| explain                                                                                       |
+-----------------------------------------------------------------------------------------------+
| Expression ((Projection + Before ORDER BY))                                                   |
|   Aggregating                                                                                 |
|     Expression ((Before GROUP BY + Projection))                                               |
|       MergingSorted (Merge sorted streams for ORDER BY)                                       |
|         SettingQuotaAndLimits (Set limits and quota after reading from storage)               |
|           Union                                                                               |
|             Expression (Convert block structure for query from local replica)                 |
|               MergingSorted (Merge sorted streams for ORDER BY)                               |
|                 MergeSorting (Merge sorted blocks for ORDER BY)                               |
|                   PartialSorting (Sort each block for ORDER BY)                               |
|                     Expression (Before ORDER BY)                                              |
|                       SettingQuotaAndLimits (Set limits and quota after reading from storage) |
|                         ReadFromStorage (MergeTree)                                           |
|             ReadFromPreparedSource (Read from remote replica)                                 |
+-----------------------------------------------------------------------------------------------+
14 rows in set (0.08 sec)

 

对比MergeTree引擎的order by执行计划,ReplicatedAggregatingMergeTree的要简化很多

优化点显而易见：分机器/分区/分段的读取数据做排序，在汇总 ， 并且是按照block进行排序的

1、对每台机器，按照一定范围进行PartialSorting（不按照具体数据排序，而是按照block快排序）

2、对第一步排序好的block进行汇总排序（因为已经知道第一步block的顺序了,所以汇总的时候直接对比最大block和最小block就可以了）

3、本地的排序好了，在做分布式的排序进行最终的汇总排序

4、求count

上面的操作很像归并排序，适应大数据量分布式排序

4、对sql做优化

【做完上一步其实已经慢满足需求了，但我们的需求并不是单纯为了求数量，而是拿到自定义时间段的某个第一个出现的东西】

切换引擎后，查询速度有了本质的提升,接下来在对sql做优化。假如原来的sql是查询3个月的数据量，那么还可以进行分段查，比如每15天/一个月 做一次order by

原来的sql：

select count(1) from 
(select 
dt,
toUInt32(dvid) as dvid ,
json 
from
表
where filter条件 and dt >= \'2021-01-01\' <= \'2021-03-31\'
order by order_key排序条件)

优化后的sql：

select count(1) from (
    select dt,toUInt32(dvid) as dvid ,json from 表 where filter条件 and dt>=\'2021-01-01\' and dt <= \'2021-01-31\' order by order_key排序条件
    union all
    select dt,toUInt32(dvid) as dvid ,json from 表 where filter条件 and dt>=\'2021-02-01\' and dt <= \'2021-02-31\' order by order_key排序条件
    union all
    select dt,toUInt32(dvid) as dvid ,json from 表 where filter条件 and dt>=\'2021-03-01\' and dt <= \'2021-03-31\' order by order_key排序条件
)

查询速度：

+---------+
| count() |
+---------+
| 3290193 |
+---------+
1 row in set (2.74 sec)

经过优化引擎+优化sql：最终的查询速度由原来的37.5s 优化2.74s