简介：数据湖概念日益炽热，本文由阿里云开源大数据 OLAP 团队和 StarRocks 数据湖剖析团队独特为大家介绍“StarRocks 极速数据湖剖析”背地的原理。
StarRocks 是一个弱小的数据分析系统，次要主旨是为用户提供极速、对立并且易用的数据分析能力，以帮忙用户通过更小的应用老本来更快的洞察数据的价值。通过精简的架构、高效的向量化引擎以及全新设计的基于老本的优化器（CBO），StarRocks 的剖析性能（尤其是多表 JOIN 查问）得以远超同类产品。
为了可能满足更多用户对于极速剖析数据的需要，同时让 StarRocks 弱小的剖析能力利用在更加宽泛的数据集上，阿里云开源大数据 OLAP 团队联结社区一起加强 StarRocks 的数据湖剖析能力。使其不仅可能剖析存储在 StarRocks 本地的数据，还可能以同样杰出的体现剖析存储在 Apache Hive、Apache Iceberg 和 Apache Hudi 等开源数据湖或数据仓库的数据。
本文将重点介绍 StarRocks 极速数据湖剖析能力背地的技术底细，性能体现以及将来的布局。
一、整体架构
在数据湖剖析的场景中，StarRocks 次要负责数据的计算剖析，而数据湖则次要负责数据的存储、组织和保护。上图描述了由 StarRocks 和数据湖所形成的实现的技术栈。
StarRocks 的架构十分简洁，整个零碎的外围只有 FE（Frontend）、BE（Backend）两类过程，不依赖任何内部组件，不便部署与保护。其中 FE 次要负责解析查问语句（SQL），优化查问以及查问的调度，而 BE 则次要负责从数据湖中读取数据，并实现一系列的 Filter 和 Aggregate 等操作。
数据湖自身是一类技术概念的汇合，常见的数据湖通常蕴含 Table Format、File Format 和 Storage 三大模块。其中 Table Format 是数据湖的“UI”，其次要作用是组织结构化、半结构化，甚至是非结构化的数据，使其得以存储在像 HDFS 这样的分布式文件系统或者像 OSS 和 S3 这样的对象存储中，并且对外裸露表构造的相干语义。Table Format 蕴含两大流派，一种是将元数据组织成一系列文件，并同理论数据一起存储在分布式文件系统或对象存储中，例如 Apache Iceberg、Apache Hudi 和 Delta Lake 都属于这种形式；还有一种是应用定制的 metadata service 来独自寄存元数据，例如 StarRocks 本地表，Snowflake 和 Apache Hive 都是这种形式。
File Format 的次要作用是给数据单元提供一种便于高效检索和高效压缩的表达方式，目前常见的开源文件格式有列式的 Apache Parquet 和 Apache ORC，行式的 Apache Avro 等。
Storage 是数据湖存储数据的模块，目前数据湖最常应用的 Storage 次要是分布式文件系统 HDFS，对象存储 OSS 和 S3 等。
FE
FE 的次要作用将 SQL 语句转换成 BE 可能意识的 Fragment，如果把 BE 集群当成一个分布式的线程池的话，那么 Fragment 就是线程池中的 Task。从 SQL 文本到分布式物理执行打算，FE 的次要工作须要通过以下几个步骤:
- SQL Parse：将 SQL 文本转换成一个 AST(形象语法树)
- SQL Analyze：基于 AST 进行语法和语义剖析
- SQL Logical Plan：将 AST 转换成逻辑打算
- SQL Optimize：基于关系代数，统计信息，Cost 模型对 逻辑打算进行重写，转换，抉择出 Cost“最低”的物理执行打算
- 生成 Plan Fragment：将 Optimizer 抉择的物理执行打算转换为 BE 能够间接执行的 Plan Fragment。
- 执行打算的调度
BE
Backend 是 StarRocks 的后端节点，负责数据存储以及 SQL 计算执行等工作。
StarRocks 的 BE 节点都是齐全对等的，FE 依照肯定策略将数据调配到对应的 BE 节点。在数据导入时，数据会间接写入到 BE 节点，不会通过 FE 直达，BE 负责将导入数据写成对应的格局以及生成相干索引。在执行 SQL 计算时，一条 SQL 语句首先会依照具体的语义布局成逻辑执行单元，而后再依照数据的散布状况拆分成具体的物理执行单元。物理执行单元会在数据存储的节点上进行执行，这样能够防止数据的传输与拷贝，从而可能失去极致的查问性能。
二、技术细节
StarRocks 为什么这么快
CBO 优化器
个别 SQL 越简单，Join 的表越多，数据量越大，查问优化器的意义就越大，因为不同执行形式的性能差异可能有成千盈百倍。StarRocks 优化器次要基于 Cascades 和 ORCA 论文实现，并联合 StarRocks 执行器和调度器进行了深度定制，优化和翻新。残缺反对了 TPC-DS 99 条 SQL，实现了公共表达式复用，相干子查问重写，Lateral Join，CTE 复用，Join Rorder，Join 分布式执行策略抉择，Runtime Filter 下推，低基数字典优化 等重要性能和优化。
CBO 优化器好坏的要害之一是 Cost 预计是否精确，而 Cost 预计是否精确的关键点之一是统计信息是否收集及时，精确。StarRocks 目前反对表级别和列级别的统计信息，反对主动收集和手动收集两种形式，无论主动还是手动，都反对全量和抽样收集两种形式。
MPP 执行
MPP (massively parallel processing) 是大规模并行计算的简称，外围做法是将查问 Plan 拆分成很多能够在单个节点上执行的计算实例，而后多个节点并行执行。每个节点不共享 CPU，内存, 磁盘资源。MPP 数据库的查问性能能够随着集群的程度扩大而一直晋升。
如上图所示，StarRocks 会将一个查问在逻辑上切分为多个 Query Fragment(查问片段)，每个 Query Fragment 能够有一个或者多个 Fragment 执行实例，每个 Fragment 执行实例 会被调度到集群某个 BE 上执行。如上图所示，一个 Fragment 能够包含 一个 或者多个 Operator（执行算子），图中的 Fragment 包含了 Scan, Filter, Aggregate。如上图所示，每个 Fragment 能够有不同的并行度。
如上图所示，多个 Fragment 之间会以 Pipeline 的形式在内存中并行执行，而不是像批处理引擎那样 Stage By Stage 执行。
如上图所示，Shuffle（数据重散布）操作是 MPP 数据库查问性能能够随着集群的程度扩大而一直晋升的要害，也是实现高基数聚合和大表 Join 的要害。
向量化执行引擎
随着数据库执行的瓶颈逐步从 IO 转移到 CPU，为了充分发挥 CPU 的执行性能，StarRocks 基于向量化技术从新实现了整个执行引擎。算子和表达式向量化执行的外围是批量按列执行，批量执行，相比与单行执行，能够有更少的虚函数调用，更少的分支判断；按列执行，相比于按行执行，对 CPU Cache 更敌对，更易于 SIMD 优化。
向量化执行不仅仅是数据库所有算子的向量化和表达式的向量化，而是一项微小和简单的性能优化工程，包含数据在磁盘，内存，网络中的按列组织，数据结构和算法的从新设计，内存治理的从新设计，SIMD 指令优化，CPU Cache 优化，C++ 优化等。向量化执行相比之前的按行执行，整体性能晋升了 5 到 10 倍。
StarRocks 如何优化数据湖剖析
大数据分析畛域，数据除了存储在数仓之外，也会存储在数据湖当中，传统的数据湖实现计划包含 Hive/HDFS。近几年比拟炽热的是 LakeHouse 概念，常见的实现计划包含 Iceberg/Hudi/Delta。那么 StarRocks 是否帮忙用户更好地开掘数据湖中的数据价值呢？答案是必定的。
在后面的内容中咱们介绍了 StarRocks 如何实现极速剖析，如果将这些能力用于数据湖必定会带来更好地数据湖剖析体验。在这部分内容中，咱们会介绍 StarRocks 是如何实现极速数据湖剖析的。
咱们先看一下全局的架构，StarRocks 和数据湖剖析相干的次要几个模块如下图所示。其中 Data Management 由数据湖提供，Data Storage 由对象存储 OSS/S3，或者是分布式文件系统 HDFS 提供。
目前，StarRocks 曾经反对的数据湖剖析能力能够演绎为上面几个局部：
- 反对 Iceberg v1 表查问 https://github.com/StarRocks/…
- 反对 Hive 表面查问 内部表 @ External_table @ StarRocks Docs (dorisdb.com)
- 反对 Hudi COW 表查问 https://github.com/StarRocks/…
接下来咱们从查问优化和查问执行这几个方面来看一下，StarRocks 是如何实现将极速剖析的能力赋予数据湖的。
查问优化
查问优化这部分次要是利用后面介绍的 CBO 优化器来实现，数据湖模块须要给优化器统计信息。基于这些统计信息，优化器会利用一系列策略来实现查问执行打算的最优化。接下来咱们通过例子看一下几个常见的策略。
统计信息
咱们看上面这个例子，生成的执行打算中，HdfsScanNode 蕴含了 cardunality、avgRowSize 等统计信息的展现。
MySQL [hive_test]> explain select l_quantity from lineitem;
+-----------------------------+
| Explain String |
+-----------------------------+
| PLAN FRAGMENT 0 |
| OUTPUT EXPRS:5: l_quantity |
| PARTITION: UNPARTITIONED |
| |
| RESULT SINK |
| |
| 1:EXCHANGE |
| |
| PLAN FRAGMENT 1 |
| OUTPUT EXPRS: |
| PARTITION: RANDOM |
| |
| STREAM DATA SINK |
| EXCHANGE ID: 01 |
| UNPARTITIONED |
| |
| 0:HdfsScanNode |
| TABLE: lineitem |
| partitions=1/1 |
| cardinality=126059930 |
| avgRowSize=8.0 |
| numNodes=0 |
+-----------------------------+ 在正式进入到 CBO 优化器之前，这些统计信息都会计算好。比方针对 Hive 咱们有 MetaData Cache 来缓存这些信息，针对 Iceberg 咱们通过 Iceberg 的 manifest 信息来计算这些统计信息。获取到这些统计信息之后，对于后续的优化策略的成果有很大地晋升。
分区裁剪
分区裁剪是只有当指标表为分区表时，才能够进行的一种优化形式。分区裁剪通过剖析查问语句中的过滤条件，只抉择可能满足条件的分区，不扫描匹配不上的分区，进而显著地缩小计算的数据量。比方上面的例子，咱们创立了一个以 ss_sold_date_sk 为分区列的表面。
create external table store_sales(
ss_sold_time_sk bigint
, ss_item_sk bigint
, ss_customer_sk bigint
, ss_coupon_amt decimal(7,2)
, ss_net_paid decimal(7,2)
, ss_net_paid_inc_tax decimal(7,2)
, ss_net_profit decimal(7,2)
, ss_sold_date_sk bigint
) ENGINE=HIVE
PROPERTIES (
"resource" = "hive_tpcds",
"database" = "tpcds",
"table" = "store_sales"
); 在执行如下查问的时候，分区 2451911 和 2451941 之间的数据才会被读取，其余分区的数据会被过滤掉，这能够节约很大一部分的网络 IO 的耗费。
select ss_sold_time_sk from store_sales
where ss_sold_date_sk between 2451911 and 2451941
order ss_sold_time_sk; Join Reorder
多个表的 Join 的查问效率和各个表参加 Join 的程序有很大关系。如 select * from T0, T1, T2 where T0.a=T1.a and T2.a=T1.a，这个 SQL 中可能的执行程序有上面两种状况：
- T0 和 T1 先做 Join，而后再和 T2 做 Join
- T1 和 T2 先做 Join，而后再和 T0 做 Join
依据 T0 和 T2 的数据量及数据分布，这两种执行程序会有不同的性能体现。针对这个状况，StarRocks 在优化器中实现了基于 DP 和贪婪的 Join Reorder 机制。目前针对 Hive 的数据分析，曾经反对了 Join Reorder，其余的数据源的反对也正在开发中。上面是一个例子：
MySQL [hive_test]> explain select * from T0, T1, T2 where T2.str=T0.str and T1.str=T0.str;
+----------------------------------------------+
| Explain String |
+----------------------------------------------+
| PLAN FRAGMENT 0 |
| OUTPUT EXPRS:1: str | 2: str | 3: str |
| PARTITION: UNPARTITIONED |
| RESULT SINK |
| 8:EXCHANGE |
| PLAN FRAGMENT 1 |
| OUTPUT EXPRS: |
| PARTITION: HASH_PARTITIONED: 2: str |
| STREAM DATA SINK |
| EXCHANGE ID: 08 |
| UNPARTITIONED |
| 7:HASH JOIN |
| | join op: INNER JOIN (BUCKET_SHUFFLE(S)) |
| | hash predicates: |
| | colocate: false, reason: |
| | equal join conjunct: 1: str = 3: str |
| |----6:EXCHANGE |
| 4:HASH JOIN |
| | join op: INNER JOIN (PARTITIONED) |
| | hash predicates: |
| | colocate: false, reason: |
| | equal join conjunct: 2: str = 1: str |
| |----3:EXCHANGE |
| 1:EXCHANGE |
| PLAN FRAGMENT 2 |
| OUTPUT EXPRS: |
| PARTITION: RANDOM |
| STREAM DATA SINK |
| EXCHANGE ID: 06 |
| HASH_PARTITIONED: 3: str |
| 5:HdfsScanNode |
| TABLE: T2 |
| partitions=1/1 |
| cardinality=1 |
| avgRowSize=16.0 |
| numNodes=0 |
| PLAN FRAGMENT 3 |
| OUTPUT EXPRS: |
| PARTITION: RANDOM |
| STREAM DATA SINK |
| EXCHANGE ID: 03 |
| HASH_PARTITIONED: 1: str |
| 2:HdfsScanNode |
| TABLE: T0 |
| partitions=1/1 |
| cardinality=1 |
| avgRowSize=16.0 |
| numNodes=0 |
| PLAN FRAGMENT 4 |
| OUTPUT EXPRS: |
| PARTITION: RANDOM |
| STREAM DATA SINK |
| EXCHANGE ID: 01 |
| HASH_PARTITIONED: 2: str |
| 0:HdfsScanNode |
| TABLE: T1 |
| partitions=1/1 |
| cardinality=1 |
| avgRowSize=16.0 |
| numNodes=0 |
+----------------------------------------------+ 谓词下推
谓词下推将查问语句中的过滤表达式计算尽可能下推到间隔数据源最近的中央，从而缩小数据传输或计算的开销。针对数据湖场景，咱们实现了将 Min/Max 等过滤条件下推到 Parquet 中，在读取 Parquet 文件的时候，可能疾速地过滤掉不必的 Row Group。
比方，对于上面的查问，l_discount= 1 对应条件会下推到 Parquet 侧。MySQL [hive_test]> explain select l_quantity from lineitem where l_discount=1;
+----------------------------------------------------+
| Explain String |
+----------------------------------------------------+
| PLAN FRAGMENT 0 |
| OUTPUT EXPRS:5: l_quantity |
| PARTITION: UNPARTITIONED |
| |
| RESULT SINK |
| |
| 2:EXCHANGE |
| |
| PLAN FRAGMENT 1 |
| OUTPUT EXPRS: |
| PARTITION: RANDOM |
| |
| STREAM DATA SINK |
| EXCHANGE ID: 02 |
| UNPARTITIONED |
| |
| 1:Project |
| | <slot 5> : 5: l_quantity |
| | |
| 0:HdfsScanNode |
| TABLE: lineitem |
| NON-PARTITION PREDICATES: 7: l_discount = 1.0 |
| partitions=1/1 |
| cardinality=63029965 |
| avgRowSize=16.0 |
| numNodes=0 |
+----------------------------------------------------+ 其余策略
除了下面介绍的几种策略，针对数据湖剖析，咱们还适配了如 Limit 下推、TopN 下推、子查问优化等策略。可能进一步地优化查问性能。
查问执行
后面介绍了，StarRocks 的执行引擎是全向量化、MPP 架构的，这些无疑都会给咱们剖析数据湖的数据带来很大晋升。接下来咱们看一下 StarRocks 是如何调度和执行数据湖剖析查问的。
查问调度
数据湖的数据个别都存储在如 HDFS、OSS 上，思考到混部和非混部的状况。咱们对 Fragment 的调度，实现了一套负载平衡的算法。
- 做完分区裁剪之后，失去要查问的所有 HDFS 文件 block
- 对每个 block 结构 THdfsScanRange，其中 hosts 蕴含 block 所有正本所在的 datanode 地址，最终失去 List
- Coordinator 保护一个所有 be 以后曾经调配的 scan range 数目的 map，每个 datanode 上磁盘已调配的要读取 block 的数目的 map>，及每个 be 平均分配的 scan range 数目 numScanRangePerBe
如果 block 正本所在的 datanode 有 be（混部）
每个 scan range 优先调配给正本所在的 be 中 scan range 数目起码的 be。如果 be 曾经调配的 scan range 数目大于 numScanRangePerBe，则从近程 be 中抉择 scan range 数目最小的
如果有多个 be 上 scan range 数目一样小，则思考 be 上磁盘的状况，抉择正本所在磁盘上已调配的要读取 block 数目小的 be
如果 block 正本所在的 datanode 机器没有 be（独自部署或者能够近程读）
抉择 scan range 数目最小的 be
查问执行
在调度到 BE 端进行执行之后，整个执行过程都是向量化的。具体看上面 Iceberg 的例子，IcebergScanNode 对应的 BE 端目前是 HdfsScanNode 的向量化实现，其余算子也是相似，在 BE 端都是向量化的实现。
MySQL [external_db_snappy_yuzhou]> explain select c_customer_id customer_id
-> ,c_first_name customer_first_name
-> ,c_last_name customer_last_name
-> ,c_preferred_cust_flag customer_preferred_cust_flag
-> ,c_birth_country customer_birth_country
-> ,c_login customer_login
-> ,c_email_address customer_email_address
-> ,d_year dyear
-> ,'s' sale_type
-> from customer, store_sales, date_dim
-> where c_customer_sk = ss_customer_sk
-> and ss_sold_date_sk = d_date_sk;
+------------------------------------------------
| PLAN FRAGMENT 0
| OUTPUT EXPRS:2: c_customer_id | 9: c_first_name | 10: c_last_name | 11: c_preferred_cust_flag | 15: c_birth_country | 16: c_login | 17: c_email_address | 48: d_year | 70: expr |
| PARTITION: UNPARTITIONED
| RESULT SINK
| 9:EXCHANGE
| PLAN FRAGMENT 1
| OUTPUT EXPRS:
| PARTITION: RANDOM
| STREAM DATA SINK
| EXCHANGE ID: 09
| UNPARTITIONED
| 8:Project
| | <slot 2> : 2: c_customer_id
| | <slot 9> : 9: c_first_name
| | <slot 10> : 10: c_last_name
| | <slot 11> : 11: c_preferred_cust_flag
| | <slot 15> : 15: c_birth_country
| | <slot 16> : 16: c_login
| | <slot 17> : 17: c_email_address
| | <slot 48> : 48: d_year
| | <slot 70> : 's'
| 7:HASH JOIN
| | join op: INNER JOIN (BROADCAST)
| | hash predicates:
| | colocate: false, reason:
| | equal join conjunct: 21: ss_customer_sk = 1: c_customer_sk
| 4:Project
| | <slot 21> : 21: ss_customer_sk
| | <slot 48> : 48: d_year
| 3:HASH JOIN
| | join op: INNER JOIN (BROADCAST)
| | hash predicates:
| | colocate: false, reason:
| | equal join conjunct: 41: ss_sold_date_sk = 42: d_date_sk
| 0:IcebergScanNode
| TABLE: store_sales
| cardinality=28800991
| avgRowSize=1.4884362
| numNodes=0
| PLAN FRAGMENT 2
| OUTPUT EXPRS:
| PARTITION: RANDOM
| STREAM DATA SINK
| EXCHANGE ID: 06
| UNPARTITIONED
| 5:IcebergScanNode
| TABLE: customer
| cardinality=500000
| avgRowSize=36.93911
| numNodes=0
| PLAN FRAGMENT 3
| OUTPUT EXPRS:
| PARTITION: RANDOM
| STREAM DATA SINK
| EXCHANGE ID: 02
| UNPARTITIONED
| 1:IcebergScanNode
| TABLE: date_dim
| cardinality=73049
| avgRowSize=4.026941
| numNodes=0
三、基准测试
TPC-H 是美国交易解决效力委员会 TPC（Transaction Processing Performance Council）组织制订的用来模仿决策反对类利用的测试集。It consists of a suite of business oriented ad-hoc queries and concurrent data modifications.
TPC-H 依据实在的生产运行环境来建模，模仿了一套销售零碎的数据仓库。该测试共蕴含 8 张表，数据量可设定从 1 GB~3 TB 不等。其基准测试共蕴含了 22 个查问，次要评估指标为各个查问的响应工夫，即从提交查问到后果返回所需工夫。
测试论断
在 TPCH 100G 规模的数据集上进行比照测试，共 22 个查问，后果如下：
StarRocks 应用本地存储查问和 Hive 表面查问两种形式进行测试。其中，StarRocks On Hive 和 Trino On Hive 查问的是同一份数据，数据采纳 ORC 格局存储，采纳 zlib 格局压缩。测试环境应用阿里云 EMR 进行构建。
最终，StarRocks 本地存储查问总耗时为 21s，StarRocks Hive 表面查问总耗时 92s。Trino 查问总耗时 307s。能够看到 StarRocks On Hive 在查问性能方面远远超过 Trino，然而比照本地存储查问还有不小的间隔，次要的起因是拜访远端存储减少了网络开销，以及远端存储的延时和 IOPS 通常都不如本地存储，前面的打算是通过 Cache 等机制补救问题，进一步缩短 StarRocks 本地表和 StarRocks On Hive 的差距。
具体测试过程请参考：StarRocks vs Trino TPCH 性能测试比照报告
四、将来布局
得益于全面向量化执行引擎，CBO 优化器以及 MPP 执行框架等核心技术，目前 StarRocks 曾经实现了远超其余同类产品的极速数据湖剖析能力。从久远来看，StarRocks 在数据湖剖析方向的愿景是为用户提供极其简略、易用和高速的数据湖剖析能力。为了可能实现这一指标，StarRocks 当初还有许多工作须要实现，其中包含：
集成 Pipeline 执行引擎，通过 Push Based 的流水线执行形式，进一步升高查问响应速度
主动的冷热数据分层存储，用户能够将频繁更新的热数据存储在 StarRocks 本地表上，StarRocks 会定期主动将冷数据从本地表迁徙到数据湖
去掉显式建设表面的步骤，用户只须要建设数据湖对应的 resource 即可实现数据湖库表全自动同步
进一步欠缺 StarRocks 对于数据湖产品个性的反对，包含反对 Apache Hudi 的 MOR 表和 Apache Iceberg 的 v2 表；反对间接写数据湖；反对 Time Travel 查问，欠缺 Catalog 的反对度等
通过层级 Cache 来进一步晋升数据湖剖析的性能
五、更多信息
参考链接
[1] https://help.aliyun.com/docum…
[2] https://github.com/StarRocks/…
[3] https://docs.dorisdb.com/zh-c…
[4] https://github.com/StarRocks/…
[5] StarRocks vs Trino TPCH 性能测试比照报告
原文链接
本文为阿里云原创内容，未经容许不得转载。