首页 > 其他分享 >浅谈 ByteHouse Projection 优化实践

浅谈 ByteHouse Projection 优化实践

时间:2023-06-06 13:01:36浏览次数:44  
标签:projection number 查询 Projection ByteHouse part 数据 浅谈

预聚合是 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 优化器的能力简单总结如下:

浅谈 ByteHouse Projection 优化实践_olap

  • 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 还没物化,可以使用原始数据进行计算。 具体改写执行逻辑:

  • 计划阶段
  1. 将原查询计划和已有 projection 进行匹配筛选能满足查询要求的 projection candidates;
  2. 基于最小的 mark 读取数选择最优的 projection candidate;
  3. 对原查询计划中的 ActionDAG 进行改写和折叠,之后用于 projection part 数据的后续计算;
  4. 将当前数据处理阶段提升到 WithMergeableState;
  • 执行阶段
  1. MergeTreeDataSelectExecutor 会将 aggregate 之前的计算进行拆分:对于 normal part,使用原查询计划进行计算;对于 projection part,使用改写后 ActionDAG 构造 QueryPipeline;
  2. 将两份数据合并,用于 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.eventsWHERE 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)

浅谈 ByteHouse Projection 优化实践_预聚合_02

  • Q1 查询计划(optimizer_projection_support=1)

浅谈 ByteHouse Projection 优化实践_预聚合_03

混合读取 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
  • Q2 执行计划
  • 按照 partition 来匹配 projection 查询过滤条件 WHERE event_date >= '2023-05-28' 会读取是三个分区的数据, 并且 agg_sum_proj_1, agg_sum_proj_2 都满足 Q2 的查询条件,所以 table scan 会读取 2023-05-28 的原始数据,2023-05-29 会读取 agg_sum_proj_1 的数据,2023-05-30 由于 agg_sum_proj_2 相对于 agg_sum_proj_1 的数据聚合度更高,读取代价较小,选择读取 agg_sum_proj_2 的数据,混合读取不同 projection 的数据。
原始表 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 数据解析与加载。

浅谈 ByteHouse Projection 优化实践_olap_04

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 存储到远端

浅谈 ByteHouse Projection 优化实践_olap_05

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 指针,对上层提供统一的访问服务。

浅谈 ByteHouse Projection 优化实践_预聚合_06

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 倍。

浅谈 ByteHouse Projection 优化实践_clickhouse_07

表结构
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;

浅谈 ByteHouse Projection 优化实践_预聚合_08

关闭 Projection

Q1

浅谈 ByteHouse Projection 优化实践_预聚合_09

Q2

浅谈 ByteHouse Projection 优化实践_预聚合_10

标签:projection,number,查询,Projection,ByteHouse,part,数据,浅谈
From: https://blog.51cto.com/bytedata/6424056

相关文章

  • 浅谈mysql索引类型(normal、unique、full textl) 的区别和使用场景
    mysql索引类型mysql索引类型normal,unique,fulltext的区别是什么?normal:表示普通索引unique:表示唯一的,不允许重复的索引,如果该字段信息保证不会重复例如身份证号用作索引时,可设置为uniquefulltextl:表示全文搜索的索引。FULLTEXT用于搜索很长一篇文章的时候,效果最好。用在......
  • 浅谈java异常[Exception]
    评:一.异常的定义在《java编程思想》中这样定义异常:阻止当前方法或作用域继续执行的问题。虽然java中有异常处理机制,但是要明确一点,决不应该用"正常"的态度来看待异常。绝对一点说异常就是某种意义上的错误,就是问题,它可能会导致程序失败。之所以java要提出异常处理机制,就是要......
  • 浅谈僵尸网络利器:Fast-flux技术——只是一些特定的apt组织才使用,倒是很少在恶意软件的
    DynamicResolution: FastFluxDNSOthersub-techniquesofDynamicResolution(3)AdversariesmayuseFastFluxDNStohideacommandandcontrolchannelbehindanarrayofrapidlychangingIPaddresseslinkedtoasingledomainresolution.Thistechniqueus......
  • 【kafka】浅谈kafka常考特性
    Kafka前几天聊完绩效的时候问了下今年还有没有涨薪,组长的原话是"很难。。。我尽量帮大家争取。。。",我刚听完脑海的第一念头:"此处涨薪难,自有不难处!"。冷静分析一波,今年整体大环境不行,还是苟着拿波年终吧,先不准备跳了,跟大家浅浅分享一下之前准备的kafka相关知识点,等看机会的时候可......
  • 浅谈业务开发与非业务开发
    软件开发在谈及业务开发与非业务开发之前,首先他们都是软件开发,那么软件开发的流程是怎样的呢?我们先来了解一下软件开发的流程。通常情况下软件开发的流程是这样的在软件开发流程中每个阶段对应相应的内容如下:需求分析:在需求分析阶段,需要明确用户的需求和期望,定义软件的功能、特性和......
  • 浅谈扫描线
    扫描线扫描线一般运用在图形上面,它和它的字面意思非常相似,就是拿一根线,在图形上面扫来扫去。我们一般用它来解决图形的面积,周长,二位数点等问题。Atlantis问题在二维坐标系上,给出多个矩形的左下以及右上坐标,求出所有矩形构成的图形的面积。我们当然知道,如果数据范围很小,我们可......
  • [浅谈] 高斯消元
    \(\color{purple}\text{P3389【模板】高斯消元法}\)所谓高斯消元就是解个\(n\)元一次方程。用矩阵记录每个方程的系数满足第\(i\)个方程:\(a[i][1]x_1+a[i][2]x_2+\dots+a[i][n]x_n=a[i][n+1]\)然后从消元,一个一个项消元,如消除\(i\)项。先选定一个此项系数绝对值最大的......
  • 分而治之--浅谈分库分表及实践之路 | 京东云技术团队
    前言之前总在聊微服务,微服务本身也是分布式系统,其实微服务的核心思想是分而治之,把一个复杂的单体系统,按照业务的交付,分成不同的自服务,以降低资深复杂度,同时可以提升系统的扩展性。今天想聊一下分库分表,因为对于快速增长的业务来说,这个是无法回避的一环。之前我在做商城相关的SAAS......
  • 浅谈字符集GB18030, GBK, GB2312, Unicode的适应性范围
    目前在中文世界里,计算机系统发展非常快速,传统的Windows已经逐渐跟不上国产化,如国产安卓系统,华为欧拉鸿蒙等系列,国产Linux系统等。国产化普遍支持GB18030!注:GB18030标准符合性认证一度属于国家强制性标准,由中国电子技术标准化研究所(CESI)认证中心进行授权认证。那么这些字符集......
  • 浅谈 Spring Bean 的生命周期
    一、Bean的生命周期概述区别于普通的Java对象需要通过new创建对象,Spring的Bean由IoC容器进行实例化、组装以及管理的。也就是说 Bean的生命周期完全由IoC容器控制。Spring容器只能管理 单例(singleton) 作用域的Bean的完整生命周期,对于 原型(prototype) 作用域......