实时数据湖数据存储模型设计

1.   数据模型

1.1.1.   Copy On Write（COW）模型

COW表模型是一种列存表，写入数据直接存储到列存Parquet文件中。对于update操作,根据主键将数据所在的文件采用最新版本重写一次，并对所有已更改的记录使用新值，该文件会有冗余历史版本存储；对于inserts操作，记录首先打包到每个分区路径中的最小文件中，直到达到配置的最大大小，之后的所有剩余记录将再次打包到新的文件组，新的文件组也会满足最大文件大小要求。

COW表支持两种读模式：

• 镜像读模式读取表内所有记录的最新状态数据，多用于批量加工
• 增量读模式仅读取变更的数据。用于增量数据加工，例如：流式加工

1.1.2.   Merge On Read（MOR）模型

MOR表模型是行列混合存储模式，日志文件采用行存，基础文件采用列存。数据记录首先会被快速的写进日志文件，后续合并操作将日志文件与基础文件合并成最新的基础文件。对于数据写入会根据主键索引判断该记录是否已经存在，该主键记录存在为更新操作，不存在为新增操作操作。数据写入分为两种模式

• 写入行存文件：Spark对更新数据写入行存文件， Flink将新增与更新都写入行存文件
• 写入列存文件：Spark将新增数据写入列存文件。

MOR表的数据读取可支持三种模式：

• 镜像读模式：读取表内所有记录的最新状态数据，多用于批量加工
• 增量读模式：读取变更的数据。用于增量数据加工，例如：流式加工
• 优化读模式：仅读取列存文件，读取的结果不一定是最新最全的数据，用户交互式查询模式，多用于对于数据准确性不高的场景，例如：探索分析场景

注：该图以Spark引擎为例，在Flink引擎下，新增数据也会写入行存log文件

1.1.3.   模型选择

从数据存储模型以及读写流程上，两类存储模型有一定的特点，在存储模式选择的时候需要根据实际业务场景进行选择。具体特点如下：

• COW表特点

COW表在数据更新的时候，会将记录所在文件的所有数据与新记录进行合并，并写入一个新的文件，因此存在一定写放大，写入性能偏低；针对全是Insert场景，如果开启小文件合并，同样也会存在合并数据重写文件的操作，因此写入性能也是受影响的，但是可以解决小文件问题。由于数据都是列存文件，且不存在数据合并读的问题，所以数据读取性能比较好。

• MOR表

在数据更新场景，数据会直接采用追加写的方式，将数据追加到行存文件，所以写入性能好，在全部是新增场景下，如果是Flink引擎也是直接追加写入log文件，性能也比较好，但是Spark引擎下也会存在小文件合并操作，写入性能偏低。由于MOR表在数据读取的时候，涉及到多版本数合并操作，读取性能偏低。另外MOR表要定期进行数据合并压缩，将历史的行存数据转为列存，通过合并操作，可以降低数据存储空间，并提升数据读取性能。

基于以上两种存储模型的特点，在选取数据模型上主要从数据读写性能方面进行考虑。典型的场景分为如下几类：

• 场景一：流式加工场景

流式场景主要应用与数据的增量读，且写入性能要求高，因此该场景建议选用MOR表。

• 场景二：批量加工

批量加工侧重于数据大吞吐，对读写性能敏感度不高，因此可以选用COW表。

• 场景三：实时入湖场景

实时入湖场景，本质是要求数据快速写入能力，因此该场景也应该采用MOR表。

• 场景四：批量入湖

生产环境，对于表都是批量入湖，不涉及后期采用实时入湖，可以采用COW表。

• 场景五：交互式分析场景

数据入湖后或者加工的结果表，需要支持交互式分析，可以采用两种方案，方案一：采用COW表，适用于批量加工，方案二：采用MOR表，适用于流式加工，但是为了提升交互式分析性能，可以将数据合并频率加高，因为合并后的数据会是列存文件读取性能高。

2.   表分区

2.1. 表分区介绍

表分区是从数据存储结构上进行优化的一种方式，采用表分区以后针对Hudi表会有以下好处：

• 数据的存储管理

采用表分区后，在数据文件存储上，会根据分区键建立分区子目录，这样数据存储逻辑与数据业务逻辑具有一定映射，在日后的运维管理上存在很大便利。例如：采用日期字段采用分区键，历史数据老化，可以直接基于分区目录进行数据清理。

• 数据写

针对Hudi特有的数据更新能力，在数据更新的时候，会对老数据文件进行合并和追加类操作，这样采用分区存储后，操作的老数据文件的数量会有一定的控制，节省写入耗费的资源，同样也会提升性能。

• 数据读

若是没有分区，在数据全量读取的时候会扫描整个表内容，耗费了很多资源处理没有必要的数据。在增加表分区后，可以进行分区裁剪，过滤掉不要扫描的分区，这样处理的数据量会大幅下降，资源消耗与性能都会改善。

分区表从存储级别上看，分为单级分区和多级分区。从分区生成方式看分为静态分区和动态分区。

2.2. 单级分区&多级分区

在指定分区键的时候，根据分区键的个数确定分区级数，单个字段代表单级分区，多个字段代表多级分区。在多级分区中，目录层级与分区键指定顺序直接相关。例如：

一级分区为dt，二级分区为hh。

在多级分区设计中，该如何设置级别，是根据我们读写的数据的业务情况来定，多级分区也就是在分区裁剪的时候会进行多轮裁剪，例如：日期分区与部门分区如何设置级别。业务场景一：多数按照部门和按照日期分析场景，没有单独按照日期分析场景，这样应该按照部门做一级分区，日期做二级分区。

2.3. 静态分区&动态分区

静态分区与动态分区的主要区别在于静态分区是手动指定，而动态分区是通过数据来进行判断。详细来说，静态分区的列实在编译时期，通过用户传递来决定的；动态分区只有在SQL执行时才能决定。

静态分区大都在指定分区导入数据场景中，在实际日常的作业运行中，应用不广。大部分采用动态分区，在建表中指定分区键，根据数据内容动态将数据写入对应的分区中。

2.4. 分区设计

基于整体表分区的能力以及基于Hudi的新特点（更新和删除数据能力），在设计分区的时候，有以下建议参考。

建议一：分区设计可以沿用Hive表分区的设计思路，根据数据业务的特点设计单级或者多级分区。

建议二：结合文件大小因素，规避因为分区粒度太小导致，分区内文件太小，造成小文件过多问题。例如：天级分区数据文件是KB级别，而采用月级分区是几十MB，这样建议采用月分区，否则天级分区带来小文件，而月级分区数据文件也不会太大，对读写性能没有什么影响。

建议三：由于Hudi支持了更新和删除操作能力，如果本次操作中包含了更新与删除，要分析一下，是否会对老分区数据进行操作。在分区粒度上，建议不要一批次操作中不要对过多的分区进行操作。例如：每批次数据更新都是对近半个月内的数据操作，该场景建议采用月分区，而不是天级分区。

建议四：如果要求数据主键全Hudi表唯一的场景，有两种方案：1）采用全局索引，该方案写入性能偏低，2）分区索引，性能高，但是对分区键要有严格要求，表主键与分区键要有稳定的映射关系。例如：采用日期分区，可以选择create_date,而不是last_update_date，原因是create_date是该记录创建的时间，后期不会变。如果选择用last_update_date作为分区键而且采用了全局索引，就会导致记录重复的问题。

3.   存储索引

3.1.1.   Hudi表索引介绍

Hudi表当前版本（0.11及之前）的索引主要用于数据在写入时，进行判断是否为update操作以及数据文件定位用，并不是传统数据库的读加速作用。后续版本Hudi会推出读加速索引。

Hudi当前索引是在表分区之下的一种数据内容与数据文件映射组织方式，也就是在使用过程中可以与表分区结合使用。

目前Hudi索引从作用域划分为两大类：全局索引和分区内索引。

• 全局索引：是指全表范围内进行索引，可保证数据全局唯一，但是由于作用域扩大，性能也会下降。
• 分区内索引：是指在表分区内进行唯一性保证。当然针对非分区表，作用域也是全表。相对于全局索引性能会比较高。

当前hudi具体支持的索引类型包括：

• HoodieBloomIndex：可以用于检索一个元素是否在一个集合中。它的优点是空间效率和查询时间都远远超过一般的算法，主要缺点是存在一定的假阳性：当其判断元素存在时，实际上元素可能并不存在。而当判定不存在时，则元素一定不存在，因此在Hudi采用BloomIndex的时候，为了规避假阳性问题，会对判断记录存在的文件再进行一遍文件内容的排查，确定是否真的存在，这样就会对性能带来一定的影响。该类索引有全局索引HoodieGlobalBloomIndex。
• Bucket索引：是一种Hash分配方式，根据指定的索引字段，计算hash值，然后结合Bucket个数，均匀分配到具体的文件中。该方式相比于BloomIndex在元素定位性能高很多，缺点是Bucket个数无法动态扩展。另外Bucket不适合于COW表，否则会导致写放大更严重。
• Simple索引：当前Hudi的默认索引，可在数据量不大的场景下使用，该方式会在每次数据写入的时候，根据目标的数据内容来判断是否为更新操作。在表数据量不大而且计算资源不紧张场景下，首选该场景。
• HbaseIndex索引：该种方式是将主键信息存储在Hbase中，在数据定位的时候，会点查Hbse服务。性能表现一般，一般不使用。
• InMemoryHashIndex：该方式是将主键信息加载到内存当中，对于Spark引擎的价值不大，一般不用。
• Ranger索引：该索引类型是华为独有索引类型，该类型在进行主键分配的时候根据Range的范围来缺点，该方式可以根据数据情况划分ranger范围，效率与Bucket索引一样。
• 状态索引：该索引是Flink的独有索引，由于Flink主要用于流式计算，主键数据会加载到flink taskmanger内存中，该种方式效率高且可以做到全局唯一。缺点是耗费内存资源，针对该缺点可以结合Flink的状态后端，将索引数据持久化到RocksDB上。对于百亿记录内的表可以采用。

3.1.2.   索引场景选择

• 场景一：流式加工场景

基于Flink的流式加工场景，可以选择内存索引（默认索引）和Bucket索引。

• 场景二：批量加工
• 对于大表（大于10亿行）采用HoodieBloomIndex索引，
• 其他采用Simple索引
• 场景三：实时入湖场景

写入的性能要求高，建议采用Bucket索引，Bucket个数=分区内数据大小/2GB.

注意：1）分区数据大小是原始数据大小不是parquet文件的大小，因为列存压缩数据文件大小不反应数据大小。2）数据个数要留有一定余量，防止数据增长后，导致单个bucket数据量过大。

• 场景四：批量入湖

批量入湖仅作为实时入湖的初始化阶段，索引与实时入湖保持一致。

若批量入湖是一种常态入湖方式根据数据量和资源情况，选择bloomfilter、bucket、ranger索引

• 场景五：交互式分析场景

对索引类型不限制，主要由写数据逻辑确定。

4.   表管理

4.1. 数据合并

4.1.1.   功能介绍

数据合并（compaction）也叫数据压缩，是针对MOR表的特有管理操作，目的是将MOR表的log文件和parquet文件合并成新的parquet文件。在该操作过程中，仅做数据合并处理，不会清理数据文件，原有的log文件和parquet文件都继续存在。数据经过合并后有以下价值：

• 读加速，由于合并到了列存parquet文件，在读merge的时候，会将最新的parquet文件与未合并的log文件merge，处理数据量相对小很多，因此查询与批量读性能会得到加速。
• 存储压缩：列存自带按列压缩能力，对于相同值较多的列压缩效果明显，这样可以降低一定存储成本。不过要结合数据清理使用，历史版本数据才会真正清理掉。

另外MOR表长期不做数据合并，也会带来一些问题：

• 读性能下降：因为MOR表读是要做数据merge操作的，如果长期为合并，就需要将历史上的所有增量版本全部合并，会耗费非常大的计算资源，性能也会大幅下降。
• 合并操作无法完成：同上原理，增量版本过多，资源需求过大，导致无法完成合并操作。

4.1.2.   合并策略

根据MOR表的原理和合并操作，若想提升MOR表的读性能，是可以将合并频率调高，保证未合并的log文件很少，性能会与COW表持平。同时又保留了MOR表高速写入的能力。

数据合并有两种调度方式：

• 同步合并：同步合并是与写操作同步，在数据写入数据文件后，检测合并策略是否满足，满足后启动合并操作，合并完成后返回客户端写入完成。是一个串行动作，如果合并频繁就会带来写入性能降低，所以采用该方式不建议合并策略制定的不要太频繁。另外合并操作会复用写入程序的资源，针对写入数据量不大，但是基础表太大，会撑大写入程序的资源消耗。
• 异步合并：异步合并是单独拉齐一个作业进程进行数据合并与写入作业解耦，合并作业的资源是单独申请、独享的。采用异步合并的优点是计算资源根据数据情况单独分配，对写入作业性能没有影响。缺点是增加了管理成本，需要维护单独合并作业。

在合并触发控制是基于提交版本数来控制，例如：合并版本数据设置为10，hoodie.compact.inline.max.delta.commits=10，代表提交10个版本做一次合并。如果我们是5分钟一次提交，那么也就是5*10分钟=50分钟做一次合并。

在生产环境中，建议实时作业采用异步合并，批量可以采用同步也可以采用异步。

4.2. 数据清理

4.2.1.   功能介绍

数据清理（clean）是指将历史无用的增量版本数据删除，该清理主要是对过程数据清理，该表的最全的镜像数据是不受影响的。数据清理主要是为了降低存储消耗，对读写性能没有影响。数据清理发生在数据合并之后，保证增量数据合并到了最新的列存文件后在做数据清理。

4.2.2.   清理策略

数据该什么时候清理，主要取决于数据存储成本和业务的要求来衡量。在数据清理常见的场景如下：

• 场景一：批量加工场景

1）不要求数据回溯，尽快降低存储成本，数据合并完成后立即清理

2）按照回溯时间要求，指定清理策略，保证清理周期内满足业务回溯需求。

• 场景二：实时加工场景

实时加工场景中，下游作业会读表的增量版本，如果清理的太过频繁，会导致下游作业异常场景下，作业重新拉起读取不到增量数据，需要从全量数据中过滤，性能会有所下降，因此在实时场景中，清理周期一般是下游做也的异常容错周期。结合存储压力，也适合太长。增量版本数据被清理后，流作业可以从parquet文件做过滤出增量数据，该能力是华为独有能力，开源hudi会出现丢数据问题。

• 场景三：长周期保留镜像版本

针对批量加工的频次不高，且要求保留长周期的历史镜像版本的场景，Hudi相较于传统数仓有很大的优势。采用该方式保留镜像版本相较于传统数仓有很大优势，因为Hudi的镜像版本之间存储冗余数据，仅仅是变更数据，不是全表冗余的情况。因此对于需要长周期保留历史镜像数据的场景，非常有价值。例如：每天一次批作业，处理今天的新数据。然后通过数据提交时间可以查询出今天的最新数据，也可以通过昨天的提交时间查询昨天的最新镜像数据。而冗余的数据仅是今天的新增数据。

4.3. 版本归档

4.3.1.   功能介绍

版本归档（archiive）在Hudi中是一个比较很轻量的操作，仅仅是对每次提交增量版本元数据进行归档，数据文件不做处理。

4.3.2.   归档策略

 在数据清理后，将对应的提交版本元数据进行归档。