湖仓存储系统设计剖析和性能优化

一、湖仓系统 阿里云EMR湖仓系统

相较于传统的数仓、数据湖来讲，湖仓系统是一种新的数据治理系统。上图展现了阿里云EMR湖仓系统的全体架构，它是围绕着Delta Lake、Iceberg、Hudi等开源数据湖格局构建的，它同时具有数仓的高性能和数据湖的低老本、放开性。这些数据湖格局基于开源的Parquet和ORC构建，能够在AWS S3、阿里OSS等低老本存储系统上运转，它还具有ACID事务、批流一体以及Upsert等才干，可以对接多种商业或开源的查问计算引擎。这些才干使得湖仓体系逐渐成为了一种趋向。

湖仓系统有必定的学习老本，比如正当性能、小文件、清算战略、性能调优等等。上方将从湖仓系统设计上入手，了解三种格局的差异。以Spark计算引擎为例，去剖析读写计算环节中的一些重要的链路和影响性能的关键点。

二、外围设计

Delta Lake、Iceberg、Hudi三个数据湖格局在性能、个性、支持水平上基本分歧，然而在详细设计上各无利害和掂量，这些设计构成了撑持湖格局个性的基石，下文将重要剖析元数据、MOR读取这两块外围设计。

元数据由schema、性能、有效的数据文件列表三个重要部分构成。传统数仓系统有独自服务来治理原数据和事务，三个数据湖格局都是将自己的元数据以自定义的数据结构耐久化到了文件系统中，搁置在表的门路下，但又和表数据分开存储。Hive表门路或分区门路下的一切数据文件都是有效的，而数据湖格局引入了多版本的概念，所以版本的有效数据文件列表须要从元数据中挑选出来。三个湖格局都封装了自身元数据的加载和降级的才干，这些可以繁难的嵌入到不同的引擎，由各个引擎Plan和Execute自己的查问。

以Spark为例来看看Delta Lake的元数据设计，它的元数据算是三个系统中最繁复的：每次对Delta Lake的写操作，或许增加字段等DDL操作，会生成一个新版本的json deltalog文件，这外面会记载元数据的变卦，蕴含一些schema的性能和file的消息，屡次commit之后，会智能发生一个checkpoint的parquet文件，这个parquet文件会包括前面一切版本的元数据消息，用于优化查问加载。

Delta Lake元数据加载流程：

Iceberg也有一个一致的元数据集，与Delta Lake不同的是，Iceberg是三层的架构。

其中metadata文件很像Delta Lake的checkpoint文件，蕴含了所有的消息，然而不同的是，metadata文件还包括了前几个快照的消息，并且Iceberg是三层架构，其manifest file能够对部分的数据文件做统计消息搜集，因此也能用于分区之下、文件之上的裁剪。

Iceberg元数据加载流程：

Hudi和前面两个很不一样，其一是它没有一致的元数据结构，其二Hudi会对数据文件启动分组，并对文件名启动编码，这是Hudi特有的file group概念。Hudi的数据必定有主键，主键可以映射到一个file group，后续关于这些主键一切的降级，都会写到这个file group，直到显式的调用修正表的文件规划。也就是说，一个file group随着屡次的commit，会发生多个版本。失掉有效数据文件列表时，会先列出分区下的一切文件，依照file group分组，取出每个group的最新文件，再依照timeline挑选掉曾经被删除的group，最后失掉一个有效的文件列表。

Hudi元数据加载流程：

Delta Lake、Iceberg、Hude元数据对比如下：

在普通状况下，在降级一个Copy-On-Write表时，即使咱们只想口头一条降级操作，也须要将一切触及到的数据文件加载出去，而后运行降级表白式，再将一切数据一同写出，这里就蕴含一些没有降级的数据，这就是写加大的现象。为了处置写加大现象，三个数据湖格局中Hudi第一个成功了Merge-On-Read表。

MOR的设计思维是，只耐久化须要写出的数据，再经过某种形式标识出来，原来的数据文件里的数据成为过时数据，读的时刻启动兼并；为了提高效率，会活期启动兼并（Compaction），通常是依照Copy-On-Write的形式写一遍。不同的MOR成功的写入、兼并战略会有所不同。

Hudi定义了一个filegroup概念，每个group包括最多1个原始数据文件和多个日志文件，数据文件运转不存在。

数据经过主键映射到filegroup，假设降级数据将会追加写到映射到filegroup内的日志文件，假设是删除，则只要要做一个主键记载，在兼并的时刻，首先读取原始的数据，而后依照这部分数据的主键去判别在增量日志中有没有相关的记载，假设存在就做兼并。

Delta Lake经过Deletion Vector的设计处置写加大的疑问。Delta Lake将须要降级、删除的数据在原数据文件中的offset标识出来，写入一个辅佐文件。Iceberg V2表有两个MOR的成功，其中基于position的设计和Delta Lake的DV是基本一样，仅在详细成功上有些区别。DV的写入仅两步：1）依据update或许delete的condition，找到文件中婚配的记载，记载他们在文件中的offset。耐久化到一个bin文件中；2）将降级后的数据，写到普通的一个新文件中。

相较于Hudi准许存在多个日志文件， Delta Lake在查问性能做了掂量，一个普通数据文件只准许随同至少一个DV文件，当对一个已存在DV文件的数据文件再做一个降级的时刻，最终写出时会把两个DV兼并的。因为DV文件中offset消息是经过位图（RoaringBitMap）来保留的，兼并操作是比拟高效的。另外Hudi是将降级的数据也写入日志文件，Delta Lake是间接写入普通的parquet文件，而后在bin文件中做一个标志。以及hudi日志文件是行存格局，Delta Lake的DV驳回自定义的格局，而数据经常使用的parquet的列存。

Delta Lake在DV下的查问形式咱们可以间接看LogicalPlan，会愈加明晰，行将原本的DeltaScan转换成Project + Filter + Delta Scan的组合。在Parquet Scan某个数据文件时，追加了_skip_row的辅佐字段，下层运行_skip_row = false的过滤，而后经过Project的投影保证仅了无辅佐字段的额外输入。

显然外围就是_skip_row的标志，DeltaLake自定义了ParquetFileFormat，在读取parquet文件后，对每个数据判别roaringBitMap能否蕴含该offset，有标志为true，没有就是false。这样就成功了Deletion Vector形式下DeltaLake表的查问。

Delta Lake、Hudi、Iceberg的MOR成功对比：

基于offset或许position的MOR成功，因为会经过扫描文件来确定位置，因此写性能上会慢于iceberg的equality或hudi mor的成功，而因为该方案不须要相似hash join的读时兼并战略，查问性能会好一些。

三、性能优化

1、查问

以Spark为例，一个完整的query链路如下：

其中与数据湖相关的有三点：元数据加载、优化plan、Table Scan。

（1）元数据加载

元数据加载包括失掉schema、结构fileindex等，分为单点加载和散布式加载两种。单点加载的代表有Iceberg、Delta Lake-Standalone，实用于小表，有内存压力。散布式加载的成功有Hudi、Delta Lake，实用于大表，需提交Spark Job。

这里咱们给出LHBench做了一个测试结果：经常使用的是TPC-H的store-sales表，设置的filesize都是10MB，单个图表内表的文件数量从1千到20W，三个图表对应的仅读取一行，读取一个分区，和普通字段作为过滤条件的三个场景。从趋向上三个场景是分歧的。咱们剖析第一个，最左侧小表Iceberg的单点形式要稍好些，但与Delta Lake的散布式元数据加载差距不大。最右侧单点形式整个Query的口头期间中蓝色口头部分基本分歧，很显著被白色startup部分限度，这部分在plan是分歧的状况下可同等于元数据加载的期间。可见，如何智能的选用适宜的加载形式，是一个可选的优化方向。

以下是两个EMR的优化案例。

案例1： EMR Manifest，是一个无服务化的优化元数据加载方案。

阿里云EMR一客户的外围ODS表，经过Spark Streaming写入Delta Lake，每天增量数据3TB，目前全表2.2PB，1500万个数据文件，仅元数据10GB。在反常状况下经常使用Hive/Presto查问，经常使用的Delta-Standalone的单机加载形式，会齐全卡住或许须要超高内存。

该优化方案是将数据文件的元数据依照分区结构提早耐久化到一个manifest文件，同时记载manifest的元数据版本。在用户查问的时刻，依据filter做分区裁剪，间接去读分区上方的manifest文件，解析出本次查问的有效数据文件，跳过了一切元数据加载的步骤。假设emr_manifest的版本有滞后，咱们也会拿到滞后的元数据，兼并失掉正确的数据快照。另外manifest文件中还会保留一些size、stats这些消息，会运行于一些文件级别的data-skipping优化。该方案在该表体量仅为300TB时提供，过后须要10GB内存90s加载完整的元数据，优化后可以成功秒级前往，且内存不须要额内查整。

案例2： EMR> 这里不得和睦HMS做一个对比，HMS仅存储表的分区级消息，查问普通表时依据分区位置去list门路，拿到有效的数据文件列表。而数据湖格局具有多版本概念，所以针对湖格局的Metastore设计必定到文件级别。另外Hudi、Iceberg基于散布式锁来成功事务性，而Delta则是基于所在文件系统的原子性和耐久性，在某些场景下不可提供更强的分歧性保证，这也是咱们成功DataLake Metastore的另外一个要素。

EMR>（2）优化plan

数据湖的第二点优化是plan优化，在前面的Spark查问例子里，调用Spark sql内核联合各种的形态统计消息去优化logic plan。比如依据运转时的一些统计消息，经过AQE去扭转join的形式，基于表的静态消息和一些cost模型来优化的CBO等。

统计消息只要表级别和column级别两类，其中表级别有statistics消息包括bytes、rows等消息，字段级别有min、max等。

关于一个繁难的join，仅经常使用表级别的rowcount消息就能够去调整join顺序，使得全体的plan环节中须要join的数据量到达起码。然而假设sql带有aggregate或许filter，就须要联合列的消息去预算整个scan的cost。

关于经过统计目的来优化查问，有几下几点思索：

在完整Logical Plan的优化并转化为Phyiscal Plan后，下一步就要口头详细的数据文件读取，即Table Scan阶段。

这里小文件对查问的影响是大家所相熟的。三个湖格局都提供了相应的兼并小文件的性能，关键疑问其一是目的文件设置多大是适宜的，其二兼并操作口头的机遇和形式，这点咱们后续再开展。自动状况下parquet在各引擎都是128M左右，但这能否是最优的还要看表的全体规模。过多的文件，将会造成元数据规模变大，影响元数据的读写，databricks和阿里云emr都提供了智能调教file size的性能。这里给出databricks的设置文件大小和表规模的映射相关。为了控制元数据的规模，关于10TB的表倡导文件大小设置为1G。

在失掉表的所有有效数据文件后，咱们还是要依据查问条件来尽或许的启动裁剪，以缩小最终读取Parquet或许ORC文件的体量。Table scan优化分为两类：一是元数据裁剪，二是行级索引。元数据裁剪又可以分为分区裁剪、Manifest裁剪（Iceberg特有）、File裁剪等不同粒度。

三个数据湖格局都支持Z-Order和DataSkipping：先将min-max消息写到元数据当中，而后联合查问过滤条件去成功过滤。

针对点查，传统数据库更多是经过行级索引来成功。Databricks商业版Delta Lake支持BloomFilter，经常使用独自的目录文件，保留数据文件的索引消息；Hudi也支持Multi-Modal Index多模索引。EMR也方案在DataLake Metastore中嵌入Index System来支持点查减速。

最后就是实践的Reader来读取数据文件了。以Parquet为例，各引擎集成parquet之后，读写性能曾经十分不错。然而有一些详细的湖格局场景会封锁一些优化参数，相关的比如谓词下推，向量化等。另外文件的紧缩格局和紧缩比也会影响文件的加载，以及目前一些Native框架支持的Native Parquet Reader（如Arrow，Presto，Velox等）。

2、写入

以一个带where条件的update SQL为例，首先要能找到婚配这些查问条件的所在的数据文件，而后加载文件，对婚配到的数据运行表白式成功降级，并最终写出。在成功提交后，兴许还要口头一些表的table service，包括clean、checkpoint、compaction等。

优化写操作，选用适宜场景的表类型（如MOR表），并性能适宜的参数。在湖格局场景下，这种后置的追加在湖格局写入前面的table service其实是逐渐成为了影响效率或许全体作业稳固性的关键性要素。要处置这样的疑问，比拟通用的做法就是拆成两个链路：一个链路就是反常写入，另外一个链路是用离线的形式经过调度一个作业来口头table service义务，包括像clean、Hudi表治理等等。

在阿里云EMR场景下，有些同时经常使用Flink和Spark引擎的客户，会驳回Flink Hudi入湖和Spark离线治理的方案。另外EMR也提供了中心化的智能化湖表治理，是联合阿里云的DLF服务虚现的。如图所示，湖格局智能将commit消息同步到DLF湖表服务，该服务将依据预约义的战略和实践表的形态及性能判别能否要发生湖表治理类的操作。

这里的外围其实是采取怎样的战略来更好的治理湖表。EMR目前上线了两大类5个战略。在同一类中，又设置优先级防止多条战略的命中而形成计算资源的糜费。举个例子，大少数入湖的表是以期间分区的，DLF湖表治理能够判别到新分区的抵达，而在第一期间对已成功写入的分区口头小文件兼并或许zorder的操作，以便极速提供这部分数据的高效查问。