广州知名网站建设,做网站的公司名字,怎么做网站程序,网站域名 邮箱文章目录基本概念时间轴(TimeLine)文件布局#xff08;File Layout#xff09;Hudi表的文件结构Hudi存储的两个部分Hudi的具体文件说明索引#xff08;Index#xff09;原理索引选项全局索引与非全局索引索引的选择策略对事实表的延迟更新对事件表的去重对维度表的随机更删…
文章目录基本概念时间轴(TimeLine)文件布局File LayoutHudi表的文件结构Hudi存储的两个部分Hudi的具体文件说明索引Index原理索引选项全局索引与非全局索引索引的选择策略对事实表的延迟更新对事件表的去重对维度表的随机更删表类型Table TypesCopy On WriteMerge On ReadCOW与MOR的对比查询类型Query Types数据写写操作写流程UPSERT写流程INSERT写流程INSERT OVERWRITEKey 生成策略删除策略总结数据读Compaction基本概念
时间轴(TimeLine) Hudi的核心是维护表上在不同的即时时间instants执行的所有操作的时间轴timeline这有助于提供表的即时视图同时还有效地支持按到达顺序检索数据。一个instant由以下三个部分组成
1Instant action在表上执行的操作类型
COMMITS一次commit表示将一批数据原子性地写入一个表。CLEANS清除表中不再需要的旧版本文件的后台活动。DELTA_COMMIT增量提交指的是将一批数据原子性地写入一个MergeOnRead类型的表其中部分或所有数据可以写入增量日志。COMPACTION合并Hudi内部差异数据结构的后台活动例如:将更新操作从基于行的log日志文件合并到列式存储的数据文件。在内部COMPACTION体现为timeline上的特殊提交。ROLLBACK表示当commit/delta_commit不成功时进行回滚其会删除在写入过程中产生的部分文件。SAVEPOINT将某些文件组标记为已保存以便其不会被删除。在发生灾难需要恢复数据的情况下它有助于将数据集还原到时间轴上的某个点。
2Instant time
通常是一个时间戳例如20190117010349它按照动作开始时间的顺序单调增加。
3State
REQUESTED表示某个action已经调度但尚未执行。INFLIGHT表示action当前正在执行。COMPLETED表示timeline上的action已经完成。
4两个时间概念
Arrival time: 数据到达 Hudi 的时间commit time。Event time: record 中记录的时间。 上图中采用时间小时作为分区字段从 10:00 开始陆续产生各种 commits10:20 来了一条 9:00 的数据根据event time该数据仍然可以落到 9:00 对应的分区通过 timeline 直接消费 10:00 commit time之后的增量更新只消费有新 commits 的 group那么这条延迟的数据仍然可以被消费到。
文件布局File Layout
Hudi表的文件结构
Hudi将一个表映射为如下文件结构 Hudi存储的两个部分
元数据.hoodie目录对应着表的元数据信息包括表的版本管理Timeline、归档目录存放过时的instant也就是版本一个instant记录了一次提交commit的行为、时间戳和状态Hudi以时间轴的形式维护了在数据集上执行的所有操作的元数据
数据和hive一样以分区方式存放数据分区里面存放着Base File.parquet和Log File.log.* Hudi的具体文件说明 Hudi将数据表组织成分布式文件系统基本路径basepath下的目录结构 表被划分为多个分区这些分区是包含该分区的数据文件的文件夹非常类似于Hive表 在每个分区中文件被组织成文件组由文件ID唯一标识 每个文件组包含几个文件片FileSlice 每个文件片包含 一个基本文件(.parquet)在某个commit/compaction即时时间instant time生成的MOR可能没有多个日志文件(.log.*)这些日志文件包含自生成基本文件以来对基本文件的插入/更新COW没有 Hudi采用了多版本并发控制(Multiversion Concurrency Control, MVCC) compaction操作合并日志和基本文件以产生新的文件片clean操作清除不使用的/旧的文件片以回收文件系统上的空间 Hudi的base fileparquet 文件在 footer 的 meta 去记录了 record key 组成的 BloomFilter用于在 file based index 的实现中实现高效率的 key contains 检测。只有不在 BloomFilter 的 key 才需要扫描整个文件消灭假阳。 Hudi 的 log avro 文件是自己编码的通过积攒数据 buffer 以 LogBlock 为单位写出每个 LogBlock 包含 magic number、size、content、footer 等信息用于数据读、校验和过滤。 索引Index
原理
Hudi通过索引机制提供高效的upserts具体是将给定的hoodie key(record key partition path)与文件id文件组建立唯一映射。这种映射关系数据第一次写入文件后保持不变所以一个 FileGroup 包含了一批 record 的所有版本记录。Index 用于区分消息是 INSERT 还是 UPDATE。 Hudi 为了消除不必要的读写引入了索引的实现。在有了索引之后更新的数据可以快速被定位到对应的 File Group。上图为例白色是基本文件黄色是更新数据有了索引机制可以做到避免读取不需要的文件、避免更新不必要的文件、无需将更新数据与历史数据做分布式关联只需要在 File Group 内做合并。
索引选项
Index类型原理优点缺点Bloom Index默认配置使用布隆过滤器来判断记录存在与否也可选使用record key的范围裁剪需要的文件 效率高不依赖外部系统数据和索引保持一致性因假阳性问题还需回溯原文件再查找一遍Simple Index把update/delete操作的新数据和老数据进行join实现最简单无需额外的资源性能比较差HBase Index把index存放在HBase里面。在插入 File Group定位阶段所有task向HBase发送 Batch Get 请求获取 Record Key 的 Mapping 信息对于小批次的keys查询效率高需要外部的系统增加了运维压力Flink State-based IndexHUDI 在 0.8.0 版本中实现的 Flink witer采用了 Flink 的 state 作为底层的 index 存储每个 records 在写入之前都会先计算目标 bucket ID。不同于 BloomFilter Index避免了每次重复的文件 index 查找
注意Flink只有一种state based index和bucket_index其他index是Spark可选配置。
全局索引与非全局索引
全局索引全局索引在全表的所有分区范围下强制要求键的唯一性也就是确保对给定的键有且只有一个对应的记录。全局索引提供了更强的保证但是随着表增大update/delete 操作损失的性能越高因此更适用于小表。
非全局索引默认的索引实现只能保证数据在分区的唯一性。非全局索引依靠写入器为同一个记录的update/delete提供一致的分区路径同时大幅提高了效率更适用于大表。
从index的维护成本和写入性能的角度考虑维护一个global index的难度更大对写入性能的影响也更大所以需要non-global index。
HBase索引本质上是一个全局索引bloom和simple index都有全局选项hoodie.index.typeGLOBAL_BLOOM 和 hoodie.index.typeGLOBAL_SIMPLE
索引的选择策略
对事实表的延迟更新
许多公司会在NoSQL数据存储中存放大量的交易数据。例如共享出行的行程表、股票买卖记录的表、和电商的订单表。这些表通常一直在增长且大部分的更新随机发生在较新的记录上而对旧记录有着长尾分布型的更新。这通常是源于交易关闭或者数据更正的延迟性。换句话说大部分更新会发生在最新的几个分区上而小部分会在旧的分区。 对于这样的作业模式布隆索引就能表现地很好因为查询索引可以靠设置得当的布隆过滤器来裁剪很多数据文件。另外如果生成的键可以以某种顺序排列参与比较的文件数会进一步通过范围裁剪而减少。Hudi用所有文件的键域来构造区间树这样能来高效地依据输入的更删记录的键域来排除不匹配的文件。
为了高效地把记录键和布隆过滤器进行比对即尽量减少过滤器的读取和均衡执行器间的工作量Hudi缓存了输入记录并使用了自定义分区器和统计规律来解决数据的偏斜。有时如果布隆过滤器的假阳性率过高查询会增加数据的打乱操作。Hudi支持动态布隆过滤器设置hoodie.bloom.index.filter.typeDYNAMIC_V0。它可以根据文件里存放的记录数量来调整大小从而达到设定的假阳性率。
对事件表的去重 事件流无处不在。从Apache Kafka或其他类似的消息总线发出的事件数通常是事实表大小的10-100倍。事件通常把时间到达时间、处理时间作为首类处理对象比如物联网的事件流、点击流数据、广告曝光数等等。由于这些大部分都是仅追加的数据插入和更新只存在于最新的几个分区中。由于重复事件可能发生在整个数据管道的任一节点在存放到数据湖前去重是一个常见的需求。
总的来说低消耗去重是一个非常有挑战的工作。虽然可以用一个键值存储来实现去重即HBase索引但索引存储的消耗会随着事件数增长而线性增长以至于变得不可行。事实上有范围裁剪功能的布隆索引是最佳的解决方案。我们可以利用作为首类处理对象的时间来构造由事件时间戳和事件idevent_tsevent_id)组成的键这样插入的记录就有了单调增长的键。这会在最新的几个分区里大幅提高裁剪文件的效益。
对维度表的随机更删 正如之前提到的如果范围比较不能裁剪许多文件的话那么布隆索引并不能带来很好的效益。在这样一个随机写入的作业场景下更新操作通常会触及表里大多数文件从而导致布隆过滤器依据输入的更新对所有文件标明阳性。最终会导致即使采用了范围比较也还是检查了所有文件。使用简单索引对此场景更合适因为它不采用提前的裁剪操作而是直接和所有文件的所需字段连接。如果额外的运维成本可以接受的话也可以采用HBase索引其对这些表能提供更加优越的查询效率。
当使用全局索引时也可以考虑通过设置hoodie.bloom.index.update.partition.pathtrue或hoodie.simple.index.update.partition.pathtrue来处理 的情况例如对于以所在城市分区的用户表会有用户迁至另一座城市的情况。这些表也非常适合采用Merge-On-Read表型。
表类型Table Types
Copy On Write
在COW表中只有数据文件/基本文件.parquet,没有增量日志文件.log.*。对每一个新批次写入都将创建相应数据文件的新版本新的FileSlice新版本文件包括旧版本文件的记录以及来自传入批次的记录全量最新。
假设我们有 3 个文件组其中包含如下数据文件。 我们进行一批新的写入在索引后我们发现这些记录与File group 1 和File group 2 匹配然后有新的插入我们将为其创建一个新的文件组File group 4。 因此data_file1 和 data_file2 都将创建更新的版本data_file1 V2 是data_file1 V1 的内容与data_file1 中传入批次匹配记录的记录合并。
由于在写入期间进行合并COW 会产生一些写入延迟。但是COW 的优势在于它的简单性不需要其他表服务如压缩也相对容易调试。
Merge On Read
MOR表中包含列存的基本文件.parquet和行存的增量日志文件基于行的avro格式.log.*。
顾名思义MOR表的合并成本在读取端。因此在写入期间我们不会合并或创建较新的数据文件版本。标记/索引完成后对于具有要更新记录的现有数据文件Hudi 创建增量日志文件并适当命名它们以便它们都属于一个文件组。 读取端将实时合并基本文件及其各自的增量日志文件。每次的读取延迟都比较高因为查询时进行合并所以 Hudi 使用压缩机制来将数据文件和日志文件合并在一起并创建更新版本的数据文件。 用户可以选择内联或异步模式运行压缩。Hudi也提供了不同的压缩策略供用户选择最常用的一种是基于提交的数量。例如可以将压缩的最大增量日志配置为 4。这意味着在进行 4 次增量写入后将对数据文件进行压缩并创建更新版本的数据文件。压缩完成后读取端只需要读取最新的数据文件而不必关心旧版本文件。
MOR表的写入行为依据 index 的不同会有细微的差别
对于 BloomFilter 这种无法对 log file 生成 index 的索引方案对于 INSERT 消息仍然会写 base file parquet format只有 UPDATE 消息会 append log 文件因为 base file 已经记录了该 UPDATE 消息的 FileGroup ID。对于可以对 log file 生成 index 的索引方案例如 Flink writer 中基于 state 的索引每次写入都是 log format并且会不断追加和 roll over。
COW与MOR的对比
Trade-OffCOWMOR数据延迟高低查询延迟低高更新IO开销高(重写所有的数据文件)低(追加到日志文件即可)Parquet文件大小单个全量使用parquet文件存储占用空间小单个全量存储使用parquetavro格式存储占用空间相对大写放大高低(依赖于合并策略)
COW一般适合用在离线批处理中MOR一般适合用在实时流处理中。
查询类型Query Types
1Snapshot Queries
快照查询可以查询指定commit/delta commit即时操作后表的最新快照。
对于写时复制COW它可以替代现有的parquet表或相同基本文件类型的表同时提供upsert/delete和其他写入方面的功能可以理解为查询最新版本的Parquet数据文件。在读时合并MOR表的情况下它通过即时合并最新文件片的基本文件和增量文件来提供近实时表几分钟。
下图是COW的快照查询 2Incremental Queries
增量查询可以查询给定commit/delta commit即时操作以来新写入的数据。有效的提供变更流来启用增量数据管道。
3 Read Optimized Queries
读优化查询可查看给定的commit/compact即时操作的表的最新快照。仅将最新文件片的基本/列文件暴露给查询并保证与非Hudi表相同的列查询性能。
下图是MOR表的快照查询与读优化查询的对比 Read Optimized Queries是对Merge On Read表类型快照查询的优化。
快照查询读优化查询数据延迟低高查询延迟高合并列式的基本文件和行式的增量日志文件低只返回列式的基本文件
4不同表支持的查询类型
COWSnapshot QueriesIncremental QueriesMORSnapshot QueriesIncremental QueriesRead Optimized Queries 数据写
写操作
UPSERT默认行为数据先通过 index 打标(INSERT/UPDATE)有一些启发式算法决定消息的组织以优化文件的大小 CDC 导入INSERT跳过 index写入效率更高 Log DeduplicationBULK_INSERT写排序对大数据量的 Hudi 表初始化友好对文件大小的限制 best effort写 HFile
写流程UPSERT
1Copy On Write
先对 records 按照 record key 去重首先对这批数据创建索引 (HoodieKey HoodieRecordLocation)通过索引区分哪些 records 是 update哪些 records 是 insertkey 第一次写入对于 update 消息会直接找到对应 key 所在的最新 FileSlice 的 base 文件并做 merge 后写新的 base file (新的 FileSlice)对于 insert 消息会扫描当前 partition 的所有 SmallFile小于一定大小的 base file然后 merge 写新的 FileSlice如果没有 SmallFile直接写新的 FileGroup FileSlice
2Merge On Read
先对 records 按照 record key 去重可选首先对这批数据创建索引 (HoodieKey HoodieRecordLocation)通过索引区分哪些 records 是 update哪些 records 是 insertkey 第一次写入如果是 insert 消息 如果 log file 不可建索引默认布隆索引会尝试 merge 分区内最小的 base file 不包含 log file 的 FileSlice生成新的 FileSlice如果没有 base file 就新写一个 FileGroup FileSlice base file如果 log file 可建索引尝试 append 小的 log file如果没有就新写一个 FileGroup FileSlice base file 如果是 update 消息写对应的 file group file slice直接 append 最新的 log file如果碰巧是当前最小的小文件会 merge base file生成新的 file slicelog file 大小达到阈值会 roll over 一个新的
写流程INSERT
1Copy On Write
先对 records 按照 record key 去重可选不会创建 Index如果有小的 base file 文件merge base file生成新的 FileSlice base file否则直接写新的 FileSlice base file
2 Merge On Read
先对 records 按照 record key 去重可选不会创建 Index如果 log file 可索引并且有小的 FileSlice尝试追加或写最新的 log file如果 log file 不可索引写一个新的 FileSlice base file
写流程INSERT OVERWRITE
在同一分区中创建新的文件组集。现有的文件组被标记为 “删除”。根据新记录的数量创建新的文件组
1COW
在插入分区之前插入相同数量的记录覆盖插入覆盖更多的记录插入重写1条记录分区包含file1-t0.parquetfile2-t0.parquet。分区将添加file3-t1.parquetfile4-t1.parquet。file1, file2在t1后的元数据中被标记为无效。分区将添加file3-t1.parquetfile4-t1.parquetfile5-t1.parquet…fileN-t1.parquet。file1, file2在t1后的元数据中被标记为无效分区将添加file3-t1.parquet。file1,file2在t1后的元数据中被标记为无效。
2MOR
在插入分区之前插入相同数量的记录覆盖插入覆盖更多的记录插入重写1条记录分区包含file1-t0.parquetfile2-t0.parquet。.file1-t00.logfile3-t1.parquetfile4-t1.parquet。file1, file2在t1后的元数据中被标记为无效。file3-t1.parquet,file4-t1.parquet…fileN-t1.parquet file1, file2在t1后的元数据中被标记为无效
3优点
COW和MOR在执行方面非常相似。不干扰MOR的compaction。减少parquet文件大小。不需要更新关键路径中的外部索引。索引实现可以检查文件组是否无效类似于在HBaseIndex中检查commit是否无效的方式。可以扩展清理策略在一定的时间窗口后删除旧文件组。
4缺点
需要转发以前提交的元数据。 在t1比如file1被标记为无效我们在t1.commit中存储 “invalidFilesfile1”(或者在MOR中存储deltacommit)在t2比如file2也被标记为无效。我们转发之前的文件并在t2.commit中标记 “invalidFilesfile1, file2”或MOR的deltacommit 忽略磁盘中存在的parquet文件也是Hudi的一个新行为, 可能容易出错,我们必须认识到新的行为并更新文件系统的所有视图来忽略它们。这一点可能会在实现其他功能时造成问题。
Key 生成策略
用来生成 HoodieKeyrecord key partition path目前支持以下策略
支持多个字段组合 record keys支持多个字段组合的 parition path 可定制时间格式Hive style path name非分区表
删除策略
1逻辑删
将 value 字段全部标记为 null
2物理删
通过 OPERATION_OPT_KEY 删除所有的输入记录配置 PAYLOAD_CLASS_OPT_KEY org.apache.hudi.EmptyHoodieRecordPayload 删除所有的输入记录在输入记录添加字段_hoodie_is_deleted
总结
通过对写流程的梳理可以了解到 Apache Hudi 相对于其他数据湖方案的核心优势
写入过程充分优化了文件存储的小文件问题Copy On Write 写会一直将一个 bucket FileGroup的 base 文件写到设定的阈值大小才会划分新的 bucketMerge On Read 写在同一个 bucket 中log file 也是一直 append 直到大小超过设定的阈值 roll over。对 UPDATE 和 DELETE 的支持非常高效一条 record 的整个生命周期操作都发生在同一个 bucket不仅减少小文件数量也提升了数据读取的效率不必要的 join 和 merge。
数据读
1Snapshot读
读取所有 partiiton 下每个 FileGroup 最新的 FileSlice 中的文件
Copy On Write 表读 parquet 文件Merge On Read 表读 parquet log 文件。
2Incremantal读
https://hudi.apache.org/docs/querying_data.html#spark-incr-query
当前的 Spark data source 可以指定消费的起始和结束 commit 时间读取 commit 增量的数据集。但是内部的实现不够高效拉取每个 commit 的全部目标文件再按照系统字段 _hoodie_commit_time_ apply 过滤条件。
3Streaming读
0.8.0 版本的 HUDI Flink writer 支持实时的增量订阅可用于同步 CDC 数据日常的数据同步 ETL pipeline。Flink 的 streaming 读做到了真正的流式读取source 定期监控新增的改动文件将读取任务下派给读 task。
Compaction
1没有 base file走 copy on write insert 流程直接 merge 所有的 log file 并写 base file
2有 base file走 copy on write upsert 流程先读 log file 建 index再读 base file最后读 log file 写新的 base file
Flink 和 Spark streaming 的 writer 都可以 apply 异步的 compaction 策略按照间隔 commits 数或者时间来触发 compaction 任务在独立的 pipeline 中执行。
