大数据数据湖数据工程【免费下载链接】hudiUpserts, Deletes And Incremental Processing on Big Data.项目地址https://gitcode.com/gh_mirrors/hu/hudi点击查看免费下载导读本文基于 Apache Hudi 的 RFC-37Metadata based Bloom Index设计文档结合仓库源码系统讲解如何借助 Hudi 元数据表Metadata Table新增的bloom_filter与column_stats两个分区为写入路径上的 HoodieBloomIndex 提供基于元数据的文件裁剪File Pruning与布隆过滤器快速查找能力。读完本文你将掌握这两个索引分区的数据模型、键值编码规则、三级查找算法、写入与读取流程以及如何通过配置项启用它们来加速大规模表与全局索引场景下的 upsert 性能。背景HoodieBloomIndex 的定位与瓶颈Hudi 在写入时需要维护多种索引用来把进入的每条记录映射到正确的 File Group。最常用的记录索引是 HoodieBloomIndex它基于各 Base File如 Parquet 文件footer 中存放的布隆过滤器在写入时确定性地把记录路由到某个 File Group并区分该记录是插入insert还是更新update。然而在表规模很大、或使用全局索引global index时HoodieBloomIndex 存在明显的性能问题每次写入都要从大量数据文件中读取布隆过滤器并进行查找对于 S3 这类云对象存储从多个文件并发读取还会面临请求限流request throttling问题在布隆查找之前还需要基于 Base File footer 中的记录键 min/max 统计做文件级裁剪读取这些 footer 同样代价高昂。RFC-37 的目标就是在 Hudi 的元数据表Metadata Table中构建一套新的元数据索引——即基于元数据表的布隆索引Metadata based Bloom Index把上述性能瓶颈中需要的文件级键值范围统计与布隆过滤器都提前沉淀到元数据表中从而显著加速 Bloom Index 的查找过程。该设计在读取路径上还与会话 RFC-27 Data skipping index数据跳过索引用于提升查询性能相互重叠、互相受益。原始 HoodieBloomIndex 完成一条记录的定位需要以下四步找到所有相关的分区partitions并列出这些分区的全部数据文件文件裁剪File Pruning从所有相关数据文件的 footer 中加载记录键 min/max 信息基于这些 min/max 构建区间范围树range interval tree过滤文件并生成文件 → 候选键的映射布隆过滤器查找Bloom Filter lookup基于布隆过滤器键查找进一步裁剪文件 → 候选键映射最终在真实数据文件中执行查找为每条输入记录确定最终位置。不难看出第 1、2 步需要所有相关数据文件的_hoodie_record_key的 min/max 值以及布隆过滤器。RFC-37 的设计正是把这两类信息放入元数据表从而快速加载所需明细、做更快的查找。总体设计在元数据表中新增两个索引分区Hudi 的元数据表是一种特殊的 Hudi 表内部表本身由若干分区组成。此前元数据表已经有一个files分区用来帮助完成分区文件列表partition file listing。RFC-37 在元数据表中新增两个分区bloom_filter—— 存放文件级file level的布隆过滤器column_stats—— 存放键范围统计key range stats。选择元数据表的原因在于元数据表使用 HBase 的 HFile 作为存储格式。HFile 是一种带索引的文件格式支持类似 Map 的按键快速查找。由于这两个索引是按文件file维度存储统计信息与布隆过滤器、且查找也以文件为粒度因此可以充分利用 HFile 的快速查找能力。下面这张图给出了元数据索引的整体设计Bloom Filter 元数据索引bloom_filter 分区数据模型与 Schemabloom_filter分区会把数据表中所有 Base File 的布隆过滤器收录进来。该分区复用元数据表已有的 payload schema并通过type字段识别布隆过滤器 payload 记录。RFC 中给出的 BloomFilterMetadata schema 如下{ doc: Metadata Index of bloom filters for all data files in the user table, name: BloomFilterMetadata, type: [ null, { doc: Data file bloom filter details, name: HoodieMetadataBloomFilter, type: record, fields: [ { doc: Bloom filter type code, name: type, type: string }, { doc: Instant timestamp when this metadata was created/updated, name: timestamp, type: string }, { doc: Bloom filter binary byte array, name: bloomFilter, type: bytes }, { doc: Bloom filter entry valid/deleted flag, name: isDeleted, type: boolean } ] } ] }在仓库的 HoodieMetadata.avsc 中该 schema 已作为HoodieMetadataRecord的BloomFilterMetadata字段落地字段定义与 RFC 完全一致对应的 Java 类为HoodieMetadataBloomFilter。type字段记录布隆过滤器的类型编码timestamp记录该元数据创建/更新的 instant 时间戳bloomFilter存放二进制字节数组isDeleted标记该条目是否已删除。键Key格式布隆过滤器记录的键由分区 Base File组合编码而来。分区名与文件名先被转换为确定性的基于哈希的 ID再经 Base64 编码后拼接为键key base64_encode(concat(hash64(partition name), hash128(file name)))采用哈希 ID 的好处是对于新写入的 insert 记录可以很容易地生成键对于需要查找的更新记录也同样容易生成不需要维护任何用于反向查找的字典。哈希位数根据目标最大规模部署下的基数cardinality与碰撞概率需求来选择对哈希 ID 再做 Base64 编码则能进一步压缩这些键的磁盘占用空间。在源码中键生成逻辑位于 HoodieMetadataPayload.getBloomFilterIndexKey即public static String getBloomFilterIndexKey(PartitionIndexID partitionIndexID, FileIndexID fileIndexID) { return partitionIndexID.asBase64EncodedString() .concat(fileIndexID.asBase64EncodedString()); }其中PartitionIndexID使用 64 位哈希见 PartitionIndexID.javaFileIndexID使用 128 位哈希见 FileIndexID.java最终通过Base64CodecUtil编码。注意非分区表空分区名会被映射为特殊的非分区标识__NON_PARTITIONED__参见 HoodieTableMetadataUtil.getPartitionIdentifier。下面这张图展示了bloom_filter分区在元数据表中的形态Column Stats 元数据索引column_stats 分区数据模型与 Schemacolumn_stats分区用于存放文件列统计信息其核心价值是让记录键查找record key lookup代码路径更加高效同时它也是数据跳过data skipping详见 RFC-27能力的基础。在更新update场景下column_stats索引用于生成文件 → 候选键映射前的文件裁剪。该分区同样复用元数据表的 payload schema由type字段区分。RFC 中给出的 ColumnStatsMetadata schema 如下{ doc: Metadata Index of column statistics for all data files in the user table, name: ColumnStatsMetadata, type: [ null, { doc: Data file column statistics, name: HoodieColumnStats, type: record, fields: [ { doc: File name for which this column statistics applies, name: fileName, type: [null, string] }, { doc: Minimum value in the range. Based on user data table schema, we can convert this to appropriate type, name: minValue, type: [null, string] }, { doc: Maximum value in the range. Based on user data table schema, we can convert it to appropriate type, name: maxValue, type: [null, string] }, { doc: Total count of values, name: valueCount, type: [null, long] }, { doc: Total count of null values, name: nullCount, type: [null, long] }, { doc: Total storage size on disk, name: totalSize, type: [null, long] }, { doc: Total uncompressed storage size on disk, name: totalUncompressedSize, type: [null, long] }, { doc: Column range entry valid/deleted flag, name: isDeleted, type: boolean } ] } ] }在仓库实际落地时该 schema 有两点演进一是 record 名称变为HoodieMetadataColumnStats并额外增加了columnName字段记录该统计对应的列名二是minValue/maxValue不再是简单的 string而是采用与 ParquetStatistics实现对齐的多种类型 wrapperBooleanWrapper、IntWrapper、LongWrapper、FloatWrapper、DoubleWrapper、BytesWrapper、StringWrapper、DateWrapper、DecimalWrapper、TimeMicrosWrapper、TimestampMicrosWrapper等以保证跨文件格式的语义一致性decimal、date、timestamp 等逻辑类型都会转换为上述基础类型之一以保持排序序关系不变。完整定义见 HoodieMetadata.avsc。键Key格式column_stats记录的键由列名 分区 Base File三元组编码而来各字段的字符串名同样先转换为确定性哈希 ID 再 Base64 编码拼接方式如下key base64_encode(concat(hash64(column name), hash64(partition name), hash128(file name)))源码中对应的生成逻辑是 HoodieMetadataPayload.getColumnStatsIndexKeypublic static String getColumnStatsIndexKey(PartitionIndexID partitionIndexID, FileIndexID fileIndexID, ColumnIndexID columnIndexID) { return columnIndexID.asBase64EncodedString() .concat(partitionIndexID.asBase64EncodedString()) .concat(fileIndexID.asBase64EncodedString()); }其中列名使用 64 位哈希见 ColumnIndexID.java分区同样使用 64 位哈希文件使用 128 位哈希。哈希 ID 在 Hudi 索引查找场景下有很多理想特性但对列级 schema 变化存在影响详见下文Schema 演进章节。下面这张图展示了column_stats分区在元数据表中的形态元数据索引查找算法三级裁剪对于进入系统的 upsert 记录给定它们的键keys为其标记当前所在位置。新的索引查找算法为生成待查分区列表生成每个分区下的键列表partitions 与每个 partition 下的 keys加载文件列表对所有涉及的分区加载其全部文件列表Level 1 —— 基于column_stats索引做范围裁剪Range pruning3.1 对每条记录键基于元组(_hoodie_record_key, partition name, file path)生成 column stats 索引查找键3.2 用该键做元数据索引查找命中则取回包含 column stats 详情的值 payload3.3 基于范围比较裁剪分区及其候选文件Level 2 —— 基于bloom_filter索引做记录裁剪Record pruning4.1 从每个分区入围的候选文件中基于元组(partition name, file path)生成 bloom filter 索引查找键4.2 用该键做元数据索引查找加载 Base File 的布隆过滤器4.3 对记录键执行布隆过滤器查找生成可能存在于该 Base File的候选键集合Level 3 —— 记录验证Record validation5.1 基于上述裁剪得到的文件列表及其候选键在真实数据文件中做最终查找确认5.2 返回最终命中键的位置file id。从源码看Level 1 与 Level 2 的批量查找实现分别位于 BaseTableMetadata.getColumnStats 与 BaseTableMetadata.getBloomFilters两者都会先按规则生成索引键集合再通过getRecordsByKeys从column_stats/bloom_filters分区批量读取记录并过滤掉isDeleted的条目同时会更新lookup_meta_index_bloom_filters、lookup_meta_index_column_ranges等指标见 HoodieMetadataMetrics.java。Level 2 中布隆过滤器的逐文件探测由 HoodieMetadataBloomFilterProbingFunction 完成它按批收集分区 Base File → 候选键映射并通过hoodieTable.getMetadataTable().getBloomFilters(...)一次性批量获取布隆过滤器再调用fileBloomFilter.mightContain(recordKey)判定候选键每批最多处理 256 个文件BLOOM_FILTER_CHECK_MAX_FILE_COUNT_PER_BATCH按每个布隆过滤器约 512K、单批总加载量不超过 128MB来估算。写入流程Writer Flow每当一个新的 commit 应用到元数据表时需要同步更新上述分区流程如下Files 分区——准备新增文件的记录Column_stats 分区——准备新增记录键格式为[ColumnIndexID][PartitionIndexID][FileIndexID] ColumnStats。这需要读取 Base File 的 footer 以获取各列的 min/max 值Bloom_filter 分区——准备新增记录键格式为[PartitionIndexID][FileIndexID] BloomFilter同样需要读取 Base File 的 footer提交——将这些记录统一提交到元数据表。第 2、3 步都需要读取 Base File footer因此可以在实现中把成本摊薄只读取一次 footer同时为两个分区准备/填充记录。此外需要确保每次 commit 的 WriteStatus 中携带足够的信息发送给元数据写入器。在源码中这两个分区的记录构建分别由 HoodieTableMetadataUtil.convertFilesToBloomFilterRecords 与 convertFilesToColumnStatsRecords 完成。细节上值得注意只有 Base FileFSUtils.isBaseFile才会生成布隆过滤器记录读取布隆过滤器通过 readBloomFilter 调用文件读取器HoodieFileReader.readBloomFilter()若读取失败则跳过该文件列统计索引仅针对 Parquet 文件生成filename.endsWith(HoodieFileFormat.PARQUET.getFileExtension())通过 readColumnRangeMetadataFrom 读取 Parquet 元数据中的列统计读取单个文件失败时会优雅回退返回空列表而不是让整个任务失败对于删除的文件会生成isDeleted true的桩记录stub而不是物理删除条目。读取流程Reader Flow当新一批写入进入 Hudi 时需要为记录标记其原始文件组位置。该索引会同时利用column_stats与bloom_filter两个分区推导出记录键 → 文件名的映射具体即上文元数据索引查找算法的三级裁剪流程。Schema 演进Schema Evolution基于 HashID 的键是从输入元组确定性生成的对于列名 分区名 文件名这一元组生成的键永远是同一个。因此一张 schema 随时间变化的表会对已生成的键产生影响。好消息是最常见的 schema 演进用例——如修改列类型change of column type、新增列adding a new column——并不受影响。受影响的是一些相对少见的用例列重命名column name rename、删除列dropping a column、以及删除后又用同名添加新列adding a column with dropped name——这些情况会导致索引引用多于实际需要进而可能让索引查找在跨演进的 schema 之间匹配到过时/新增的记录。为避免查找到过时/新增的索引记录RFC 给出了两个设计选项推荐查询改写 / 结果记录集裁剪Query rewrite / Result recordset pruningSchema 演进层应引入查询改写阶段检测输入查询是否涉及演进过的 schema并视情况给查询追加额外谓词也可以基于提交时间commit time与 schema 变更时间来裁剪最终结果记录集。使输入元组集 schema 感知Making input tuple set schema aware在列名、分区名、文件路径之外再附加一个版本/标签version/tag使生成的键高度 schema 专属但该方案有性能代价查找将退化为前缀式而非定点式pointed lookup即索引查找需要返回所有版本/标签下的记录并在此基础上再做一轮裁剪。从当前仓库看该议题仍未完全闭合属 RFC 中明确标注的遗留问题之一。配置项与启用方式RFC 中提及通过配置hoodie.metadata.file_pruning.enable控制是否启用元数据表列统计裁剪在仓库实际落地中该能力由 HoodieMetadataConfig 中一系列配置控制搜索hoodie.metadata.file_pruning.enable仅出现在 RFC 文档本身未在配置类中落地实际以下方配置为准。元数据表与 Bloom 索引相关配置配置项默认值版本说明hoodie.metadata.index.bloom.filter.enablefalse0.11.0是否在元数据表中索引数据文件的布隆过滤器。启用后元数据表会新增bloom_filters分区并在索引查找阶段使用见 HoodieMetadataConfig.java#L121-L127hoodie.metadata.index.bloom.filter.file.group.count40.11.0bloom_filters分区的 file group 数量。该值控制 Base/Log 文件大小与读取并行度建议使 Base 文件控制在 1GB 以内见 HoodieMetadataConfig.java#L129-L136hoodie.metadata.index.bloom.filter.parallelism2000.11.0生成布隆过滤器索引时的并行度见 HoodieMetadataConfig.java#L138-L143hoodie.metadata.index.bloom.filter.column.list未设置0.11.0逗号分隔的列清单指定为哪些列构建布隆过滤器索引未设置时仅索引记录键见 HoodieMetadataConfig.java#L198-L203Column Stats 索引相关配置配置项默认值版本说明hoodie.metadata.index.column.stats.enablefalse0.11.0是否在元数据表中索引数据文件的列范围信息。启用后元数据表新增column_stats分区并在索引查找阶段用于文件裁剪见 HoodieMetadataConfig.java#L145-L151hoodie.metadata.index.column.stats.file.group.count20.11.0column_stats分区的 file group 数量同样建议 Base 文件控制在 1GB 以内见 HoodieMetadataConfig.java#L153-L160hoodie.metadata.index.column.stats.parallelism2000.11.0生成列统计索引时的并行度见 HoodieMetadataConfig.java#L162-L167hoodie.metadata.index.column.stats.column.list未设置索引全部列0.11.0逗号分隔的列清单仅对这些列构建列统计索引未设置则索引所有列见 HoodieMetadataConfig.java#L169-L174hoodie.metadata.index.column.stats.processing.mode.override未设置自动判定0.12.0覆盖 Column Stats 索引的读取处理模式in-memory在进程内或engine使用 Spark 等执行引擎默认根据索引大小与读取列数自动判定见 HoodieMetadataConfig.java#L179-L187hoodie.metadata.index.column.stats.inMemory.projection.threshold1000000.12.0读取 Column Stats 索引时若预期投影行数低于该阈值则尝试在内存中加载否则使用执行引擎见 HoodieMetadataConfig.java#L189-L196此外在 Spark 写入端还需要启用hoodie.bloom.index.use.metadata让 Bloom Index 在查找阶段使用元数据表。这一点在仓库测试用例中有明确体现测试 TestHoodieBloomIndex 的参数组合rangePruning、treeFiltering、bucketizedChecking、useMetadataTable同时配置了hoodie.bloom.index.use.metadata、hoodie.metadata.index.column.stats.enable与hoodie.metadata.index.bloom.filter.enableTestBloomIndexTagWithColStats 同样以该组合验证使用元数据表中的列统计与布隆过滤器来标记记录位置。实现要点接口不变与源码验证公共接口保持不变RFC 明确要求HoodieIndex的公共接口不做任何改动。核心的tagLocation接口保持不变/** * Looks up the index and tags each incoming record with a location of a file that contains * the row (if it is actually present). */ PublicAPIMethod(maturity ApiMaturityLevel.EVOLVING) public abstract HoodieDataHoodieRecordT tagLocation( HoodieDataHoodieRecordT records, HoodieEngineContext context, HoodieTable hoodieTable) throws HoodieIndexException;RFC 中计划扩展HoodieBloomIndex.explodeRecordsWithFileComparisons()使其检查新增配置RFC 中命名为hoodie.metadata.file_pruning.enable——若启用则基于元数据表列统计做键范围文件裁剪。从当前仓库实现看Spark 侧 Bloom Index 的元数据加速路径已落地于 SparkHoodieBloomIndexHelper其中通过mapPartitionsToPair(new HoodieMetadataBloomFilterProbingFunction(...))使用元数据表布隆过滤器完成 Level 2 记录裁剪。分区类型与启用判定元数据表各分区在 MetadataPartitionType 枚举中定义FILESrecord type 2分区路径files、COLUMN_STATSrecord type 3分区路径column_stats、BLOOM_FILTERSrecord type 4分区路径bloom_filters。各分区的启用判定逻辑为COLUMN_STATS由hoodie.metadata.index.column.stats.enable控制BLOOM_FILTERS由hoodie.metadata.index.bloom.filter.enable控制二者共同通过MetadataPartitionType.getEnabledPartitions(writeConfig, metaClient)汇总条件为配置启用或分区已存在见 MetadataPartitionType.java#L391-L395。payload 的构造与合并也按分区类型分派COLUMN_STATS的 payload 合并会调用mergeColumnStatsRecords做新旧记录的合并而BLOOM_FILTERS的 payload 合并则直接取新记录——因为布隆过滤器总是增量additive追加的无需与旧布隆过滤器合并见 MetadataPartitionType.java#L119-L127 与 #L155-L158。测试计划RFC 规划的测试面包括功能测试Functionality为已存在的键标记位置tag location for existing keys为不存在的键标记位置tag location for non-existing keys性能测试Performance证明基于元数据的索引确实加速 upsert 场景升级测试UpgradeRFC 标注 TODO细节待补当前仓库中 TestHoodieBloomIndex 与 TestBloomIndexTagWithColStats 已在覆盖使用元数据表列统计 布隆过滤器完成 tag location 的功能验证。落地计划Rollout/Adoption Plan0.10.0 是一个 FlagDay 发布旧元数据表会被清空并重建为新元数据表元数据索引Metadata Index特性计划在 0.10.x 版本中落地任何为后续 minor 版本提供该特性所需的准备工作例如为存量表构建新的元数据索引——RFC 提案与文档待定都需要随 0.10.0 一起推出更详细的 rollout 计划在 RFC 中标注为 TODO。需要说明的是上述版本计划是 RFC 撰写时的规划读者应以当前实际发布的功能与配置如 0.11.0 起提供的hoodie.metadata.index.bloom.filter.enable、hoodie.metadata.index.column.stats.enable等为准。小结RFC-37 通过在 Hudi 元数据表中新增bloom_filter与column_stats两个索引分区把 Bloom Index 定位记录时最昂贵的两步——键范围文件裁剪与布隆过滤器加载——从逐个读取大量数据文件 footer转变为基于 HFile 的元数据表定点快速查找从根本上缓解了大表、全局索引与云对象存储场景下的性能瓶颈。结合hoodie.metadata.index.column.stats.enable、hoodie.metadata.index.bloom.filter.enable与hoodie.bloom.index.use.metadata等配置该能力已在实际版本中落地并通过三级裁剪算法范围裁剪 → 布隆裁剪 → 记录验证在不改动HoodieIndex公共接口的前提下完成了对写入路径的加速。赞分享大数据数据湖数据工程【免费下载链接】hudiUpserts, Deletes And Incremental Processing on Big Data.项目地址https://gitcode.com/gh_mirrors/hu/hudi点击查看免费下载相关推荐Hudi 元数据索引实战解析基于 Metadata Table 的 Bloom Filter 与 Column Stats 索引RFC-37Hudi 元数据索引实战解析基于 Metadata Table 的 Bloom Filter 与 Column Stats 索引RFC 37 导读 本文以数据湖湖仓一体大数据数据存储Hudi 元数据表上的记录级索引RFC-8 Record Index 设计与实现深度解析Hudi 元数据表上的记录级索引RFC 8 Record Index 设计与实现深度解析 本文围绕 Apache Hudi 的 RFC 8Metadata数据湖湖仓一体大数据数据存储Hudi Record IndexRecord Index 记录级索引设计与实现解析基于 Metadata Table 的亿级记录索引方案Hudi Record IndexRecord Index 记录级索引设计与实现解析基于 Metadata Table 的亿级记录索引方案 导读 本文围绕大数据数据湖数据工程上一篇三个夜晚搞定 500 首歌LRCGet 批量下载同步歌词的完整实战手记下一篇Apache Druid JavaScript 扩展开发指南运行时动态扩展的七类能力与安全实践创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考