预聚合是OLAP系统中常用的一种优化手法,在经过在加载数据时就进行部分聚合核算,生成聚合后的中间表或视图,从而在查询时直接运用这些预先核算好的聚合成果,进步查询功能,完成这种预聚合办法大多都运用物化视图来完成。
ClickHouse社区完成的Projection功用类似于物化视图,原始的概念来源于Vertica,在原始表数据加载时,依据聚合SQL界说的表达式,核算写入数据的聚合数据与原始数据同步写入存储。在数据查询的过程中,假如查询 SQL 经过匹配剖析能够经过聚合数据核算得到,直接查询聚合数据减少核算开支,大幅进步查询功能。
ClickHouse Projection是针对物化视图现有问题,在查询匹配,数据一致性上扩展了运用场景:
- 支撑normal projection,依照不同列进行数据重排,关于不同条件快速过滤数据
- 支撑aggregate projection, 运用聚合查询在源表上直接界说出预聚合模型
- 查询剖析能依据查询价值,主动挑选最优Projection进行查询优化,无需改写查询
- projeciton数据存储于原始part目录下,在任一时刻针对任一数据改换操作均提供一致性保证
- 维护简略,不需别的界说新表,在原始表添加projection属性
ByteHouse 是火山引擎依据ClickHouse研制的一款剖析型数据库产品,是一同支撑实时和离线导入的自助数据剖析平台,能够对 PB 级海量数据进行高效剖析。具有真实时剖析、存储-核算分离、多级资源隔离、云上全托管服务四大特点,为了更好的兼容社区的projection功用,扩展projection运用场景,ByteHouse对Projection进行了匹配场景和架构上进行了优化。在ByteHouse商用客户功能测验projection的功能测验,在1.2亿条的实践出产数据集中进行测验,查询并发才能进步10~20倍,下面从projeciton在优化器查询改写和依据ByteHouse框架改善两个方面谈一谈现在的优化工作。
Projection运用
为了进步ByteHouse对社区有很好的兼容性,ByteHouse保留了原有语法的支撑,projection操作分为创立,删去,物化,删去数据几个操作。为了便于理解后面的优化运用行为剖析系统例子作为剖析的对象。
语法
-- 新增projection界说
ALTER TABLE [db]. table ADD PROJECTION name ( SELECT < COLUMN LIST EXPR > [ GROUP BY ] [ ORDER BY ] )
-- 删去projection界说而且删去projection数据
ALTER TABLE [db]. table DROP PROJECTION name
-- 物化原表的某个partition数据
ALTER TABLE [db.] table MATERIALIZE PROJECTION name IN PARTITION partition_name
-- 删去projection数据但不删去projection界说
ALTER TABLE [db.] table CLEAR PROJECTION name IN PARTITION partition_name
实例
CREATE DATABASE IF NOT EXISTS tea_data;
创立原始数据表
CREATE TABLE tea_data.events(
app_id UInt32,
user_id UInt64,
event_type UInt64,
cost UInt64,
action_duration UInt64,
display_time UInt64,
event_date Date
) ENGINE = CnchMergeTree PARTITION BY toDate(event_date)
ORDER BY
(app_id, user_id, event_type);
创立projection前写入 2023-05-28 分区测验数据
INSERT INTO tea_data.events
SELECT
number / 100,
number % 10,
number % 3357,
number % 166,
number % 5,
number % 40,
'2023-05-28 05:11:55'
FROM system.numbers LIMIT 100000;
创立聚合projection
ALTER TABLE tea_data.events ADD PROJECTION agg_sum_proj_1
(
SELECT
app_id,
user_id,
event_date,
sum(action_duration)
GROUP BY app_id,
user_id, event_date
);
创立projection后写入 2023-05-29 分区测验数据
INSERT INTO tea_data.events
SELECT
number / 100,
number % 10,
number % 3357,
number % 166,
number % 5,
number % 40,
'2023-05-29 05:11:55'
FROM system.numbers LIMIT 100000;
Note:CnchMergeTree是ByteHouse特有的引擎
Query Optimizer扩展Projection改写
ByteHouse优化器
ByteHouse 优化器为业界现在仅有的ClickHouse 优化器方案。ByteHouse 优化器的才能简略总结如下:
- RBO:支撑:列裁剪、分区裁剪、表达式简化、子查询解关联、谓词下推、冗余算子消除、Outer-JOIN 转 INNER-JOIN、算子下推存储、分布式算子拆分等常见的启发式优化才能。
- CBO:依据 Cascade 查找框架,完成了高效的 Join 枚举算法,以及依据 Histogram 的价值预算,对 10 表全连接等级规划的 Join Reorder 问题,能够全量枚举并寻求最优解,一同针对大于10表规划的 Join Reorder 支撑启发式枚举并寻求最优解。CBO 支撑依据规则扩展查找空间,除了常见的 Join Reorder 问题以外,还支撑 Outer-Join/Join Reorder,Magic Set Placement 等相关优化才能。
- 分布式方案优化:面向分布式MPP数据库,生成分布式查询方案,而且和 CBO 结合在一同。相对业界干流完成:分为两个阶段,首先寻求最优的单机版方案,然后将其分布式化。咱们的方案则是将这两个阶段交融在一同,在整个 CBO 寻求最优解的过程中,会结合分布式方案的诉求,从价值的角度挑选最优的分布式方案。关于 Join/Aggregate 的还支撑 Partition 属性打开。
- 高阶优化才能:完成了 Dynamic Filter pushdown、单表物化视图改写、依据价值的 CTE (公共表达式同享)。
借助ByteHouse优化器强大的才能,针对projection原有完成的几点局限性做了优化,下面咱们先来看一下社区在projection改写的详细完成。
社区Projection改写完成
在非优化器履行模式下,对原始表的聚合查询可经过 aggregate projection 加快,即读取 projection 中的预聚合数据而不是原始数据。核算支撑了 normal partition 和 projection partition 的混合查询,假如一个 partition 的 projection 还没物化,能够运用原始数据进行核算。
详细改写履行逻辑:
-
方案阶段
- 将原查询方案和已有projection 进行匹配挑选能满意查询要求的projection candidates;
- 依据最小的 mark 读取数挑选最优的 projection candidate;
- 对原查询方案中的 ActionDAG 进行改写和折叠,之后用于 projection part 数据的后续核算;
- 将当前数据处理阶段进步到 WithMergeableState;
-
履行阶段
- MergeTreeDataSelectExecutor 会将 aggregate 之前的核算进行拆分:关于 normal part,运用原查询方案进行核算;关于 projection part,运用改写后 ActionDAG 结构QueryPipeline;
- 将两份数据合并,用于 aggregate 之后的核算。
ByteHouse优化器改写完成
优化器会将查询切分为不同的plan segment分发到worker节点并行履行,segment之间经过exchange交换数据,在plan segment内部依据query plan 构建pipeline履行,以下面简略聚合查询为例,说明优化器如何匹配projection。
Q1:
SELECT
app_id,
user_id,
sum(action_duration)
FROM tea_data.events
WHERE event_date = '2023-05-29'
GROUP BY
app_id,
user_id
在履行方案阶段优化器尽量的将 TableScan 上层的 Partial Aggregation Step,Projection 和 Filter 下推到 TableScan 中,在将plan segment发送到worker节点后,在依据查询价值挑选合适projection进行匹配改写,从下面的履行方案上看,射中projection会在table scan中直接读取AggregateFunction(sum, UInt64)的state数据,比较于没有射中projection的履行方案减少了AggregaingNode的聚合运算。
Q1查询方案(optimizer_projection_support=0) | Q1查询方案(optimizer_projection_support=1) |
---|---|
![]() ![]() |
![]() ![]() |
混合读取Projection
Projection在创立之后不支撑更新schema,只能创立新的projection,但是在一些关于projection schema 改变需求频频事务场景下,需要同一个查询既能够读取旧projection也能读取新projection,所以在匹配时需要从partition维度进行匹配而不是从projection界说的维度进行匹配,混合读取不同projection的数据,这样会使查询愈加灵敏,更好的适应事务场景,下面举个详细的实例:
创立新的projection
ALTER TABLE tea_data.events ADD PROJECTION agg_sum_proj_2
(
SELECT
app_id,
sum(action_duration),
sum(cost)
GROUP BY app_id
);
写入 2023-05-30 的数据
INSERT INTO tea_data.events
SELECT
number / 10,
number % 100,
number % 23,
number % 3434,
number % 23,
number % 55,
'2023-05-30 04:12:43'
FROM system.numbers LIMIT 100000;
履行查询
Q2:
SELECT
app_id,
sum(action_duration)
FROM tea_data.events
WHERE event_date >= '2023-05-28'
GROUP BY app_id
原始表Schema更新
当对原始表添加新字段(维度或目标),对应projection 不包括这些字段,这时候为了使用projection一般情况下需要删去projection重新做物化,比较浪费资源,假如优化器匹配算法能正确处理不存在缺省字段,并运用缺省值参加核算就能够处理这个问题。
ALTER TABLE tea_data.events ADD COLUMN device_id String after event_type;
ALTER TABLE tea_data.events ADD COLUMN stay_time UInt64 after device_id;
履行查询
Q3:
SELECT
app_id,
device_id,
sum(action_duration),
max(stay_time)
FROM tea_data.events
WHERE event_date >= '2023-05-28'
GROUP BY app_id,device_id
Q3履行方案 | 默认值参加核算 |
---|---|
![]() |
从查询方案能够看出,即使agg_sum_proj_1和agg_sum_proj_2 并不包括新增的维度字段device_id,目标字段stay_time, 依然能够射中原始的partiton的projection,而且运用默认值来参加核算,这样能够使用旧的projection进行查询加快。 |
ByteHouse Projection完成
Projection是依照ByteHouse的存算分离架构进行设计的,Projecton数据由分布式存储统一进行办理,而针对projection的查询和核算则在无状况的核算节点上进行。比较于社区版,ByteHouse Projection完成了以下优势:
- 关于Projection数据的存储节点和核算节点能够独立扩展,即能够依据不同事务关于Projection的运用需求,增加存储或许核算节点。
- 当进行Projection查询时,能够依据不同Projection的数据查询量来分配核算节点的资源,从而完成资源的隔离和优化,进步查询功率。
- Projection的元数据存储非常轻量,在事务数据急剧变化的时候,核算节点能够做到事务无感知扩缩容,无需额外的Projection数据迁移。
Projection数据存储
在ByteHouse中,多个projections数据与data数据存储在一个同享存储文件中。文件的外部数据对projections内部的内容没有感知,相当于一个黑盒。当需要读取某个projection时,经过checksums里边存储的projection指针,定位到特定projection方位,完成projection数据解析与加载。
Write操作
Projection写入分为两部分,先在本地做数据写入,产生part文件存储在worker节点本地,然后经过dumpAndCommitCnchParts将数据dump到长途同享存储。
-
写入本地
- 经过writeTempPart()将block写入本地,当写完原始part后,循环经过办法addProjectionPart()将每一个projection写入part文件夹,并添加到new_part中进行办理。
-
dump到长途存储
- dumpCnchParts()的时候,依照上述的存储格局,写入完原始part中的bin和mark数据后,循环将每一个projection文件夹中的数据写入到同享存储文件中,并记载方位和巨细到checksums,如下:
- 写入header
- 写入data
- 写入projections
- 写入Primary index
- 写入Checksums
- 写入Metainfo
- 写入Unique Key Index
- 写入data footger
Merge操作
随着时刻的推移,针对同一个partition会存在越来越多的parts,而parts越多查询过滤时的价值就会越大。因而,ByteHouse在后台进程中会merge同一个partition的parts组成更大的part,从而减少part的数量进步查询的功率。
-
关于每一个要merge的part
- 关于part中的每一列,缓存对应的segments到本地
- 创立MergeTreeReaderStreamWithSegmentCache,经过长途文件buffer或许本地segments的buffer初始化
-
经过MergingSortedTransform或AggregatingSortedTransform等将sources交融成PipelineExecutingBlockInputStream
-
创立MergedBlockOutputStream
关于projection,进行如下操作
- 建立每一个projection的读取流,本地缓存buffer或许长途文件buffer
- 原始表merge过程,对parts中的projections进行merge
- 经过dumper将新的完好part存储到远端
Mutate操作
ByteHouse采用MVCC的方法,针对mutate涉及的列,新增一个delta part版别存储此次mutate涉及到的列。相应地,咱们在mutate的时候,结构projection的mutate操作的inputstream,将mutate后的projection和原始表数据一同写到同一个delta part中。
- 在MutationsInterpreter里边,经过InterpreterSelectQuery(mutation_ast)获取BlockInputStream
- projection经过block和InterpreterSelectQuery(projection.ast)重新构建
Materialize物化操作
如下图所示,依据ByteHouse的part办理方法,针对mutate操作或新增物化操作,咱们为part生成新的delta part,在下图part中,它所办理的三个projections由base part中的proj2,delta part#1中的proj1’,以及delta part#2中的proj3共同构成。当parts加载完成后,delta part#2会存储base part中的proj2的指针和delta part#1中的proj1’指针,以及自身的proj3指针,对上层提供统一的访问服务。
Worker端磁盘缓存
现在,CNCH中针对不同数据设计了不同的缓存类型
- DiskCacheSegment:办理bin和mark数据
- ChecksumsDiskCacheSegment:办理checksums数据
- PrimaryIndexDiskCacheSegment:办理主键索引数据
- BitMapIndexDiskCacheSegment:办理bitmap索引数据
针对Projection中的数据,别离经过上述的DiskCache,ChecksumsDiskCache和PrimaryIndexDiskCache对bin,mark,checksums以及索引进行缓存。
别的,为了加快Projection数据的加载过程,咱们新增了MetaInfoDiskCacheSegment用于缓存Projection相关的元数据信息。
实践事例剖析
某真实用户场景的数据集,咱们使用它对Projection功能进行了测验。
该数据集约1.2亿条,包括projection约240G巨细,测验机器 80CPU(s) / 376G Mem,配置如下:
- SET allow_experimental_projection_optimization = 1
- use_uncompressed_cache = true
- max_threads = 1
- log_level = error
- 敞开Projection查询并发度80,封闭Projection查询并发度为30
测验成果
敞开Projection后,针对1.2亿条的数据集,查询功能进步10~20倍。
QPS (敞开Projection) | QPS (封闭Projection) | |
---|---|---|
Q1 | 87.365 | 5.697 |
Q2 | 124.780 | 4.511 |
表结构
CREATE TABLE user.trades(
`type` UInt8,
`status` UInt64,
`block_hash` String,
`sequence_number` UInt64,
`block_timestamp` DateTime,
`transaction_hash` String,
`transaction_index` UInt32,
`from_address` String,
`to_address` String,
`value` String,
`input` String,
`nonce` UInt64,
`contract_address` String,
`gas` UInt64,
`gas_price` UInt64,
`gas_used` UInt64,
`effective_gas_price` UInt64,
`cumulative_gas_used` UInt64,
`max_fee_per_gas` UInt64,
`max_priority_fee_per_gas` UInt64,
`r` String,
`s` String,
`v` UInt64,
`logs_count` UInt32,
PROJECTION tx_from_address_hit
(
SELECT *
ORDER BY from_address
),
PROJECTION tx_to_address_hit (
SELECT *
ORDER BY to_address
),
PROJECTION tx_sequence_number_hit (
SELECT *
ORDER BY sequence_number
),
PROJECTION tx_transaction_hash_hit (
SELECT *
ORDER BY transaction_hash
)
)
ENGINE=CnchMergeTree()
PRIMARY KEY (transaction_hash, from_address, to_address)
ORDER BY (transaction_hash, from_address, to_address)
PARTITION BY toDate(toStartOfMonth(`block_timestamp`));
敞开Projection
Q1
WITH tx AS ( SELECT * FROM user.trades WHERE from_address = '0x9686cd65a0e998699faf938879fb' ORDER BY sequence_number DESC,transaction_index DESC UNION ALL SELECT * FROM user.trades WHERE to_address = '0x9686cd65a0e998699faf938879fb' ORDER BY sequence_number DESC, transaction_index DESC ) SELECT * FROM tx LIMIT 100;
Q2
with tx as (select sequence_number, transaction_index, transaction_hash, input from user.trades where from_address = '0xdb03b11f5666d0e51934b43bd' order by sequence_number desc,transaction_index desc UNION ALL select sequence_number, transaction_index, transaction_hash, input from user.trades where to_address = '0xdb03b11f5666d0e51934b43bd' order by sequence_number desc, transaction_index desc) select sequence_number, transaction_hash, substring(input,1,8) as func_sign from tx order by sequence_number desc, transaction_index desc limit 100 settings max_threads = 1, allow_experimental_projection_optimization = 1, use_uncompressed_cache = true;
封闭Projection
Q1
Q2
进入官方交流群,了解更多ClickHouse&ByteHouse干货