大数据数据湖数据工程【免费下载链接】hudiUpserts, Deletes And Incremental Processing on Big Data.项目地址https://gitcode.com/gh_mirrors/hu/hudi点击查看免费下载本文以 Apache Hudi 的 RFC-80 设计提案为核心系统讲解 Hudi 如何通过列族Column Family概念重新组织 MOR 桶表的存储与读写路径解决实时宽表拼接场景下 Flink 多层 Join 带来的状态后端压力问题。读者将掌握列族的存储格式、建表/改表 SQL 语法、写入与读取的底层原理以及列族级与全量 Compaction 的取舍方法。背景实时宽表拼接的痛点在流式计算中表被加宽table widened是极为常见的场景。目前主流的实时宽表拼接方式是通过 Flink 的多层 Join 完成的。然而 Flink 的 Join 会在状态后端state backend中缓存大量数据随着数据集增大Flink 任务状态后端的压力会逐渐上升极端情况下甚至可能导致任务不可用在多层级联 Join 场景中该问题会进一步放大每个 Join 节点都需要维护自身的关联状态。Hudi 1.x 引入了将部分更新partial update编码进 log 文件的能力这本身已经可以应对上述场景但即便使用部分更新列族Column Family机制仍然能显著降低 Compaction 阶段产生的写放大write amplification。这正是 RFC-80 提出的核心动机。核心概念从 FileGroup 到 Column Family当前 Hudi 以fileGroup为粒度组织数据。RFC-80 在 fileGroup 之下进一步引入column cluster从而衍生出columnFamily概念形成如下分层结构分区内的数据按哈希被划分为多个桶bucket每个桶映射到一个 file group桶内文件进一步按 columnFamily 划分一个桶下的多个 colFamily 文件共同构成一个完整的 fileGroup当桶内只有一个 columnFamily 时表自动退化为原生 Hudi 桶表native bucket table。从设计上看列族是 Hudi 1.0 非阻塞并发写入non-blocking concurrent writing能力之上的一层组织优化可参考 WriteConcurrencyMode.java 中NON_BLOCKING_CONCURRENCY_CONTROL的说明多 writer 可同时写入同一 file group冲突由查询端 reader 与 compactor 自动消解。引入列族后的收益写性能Write performance写入过程与普通桶写入类似只是多了拆分列簇 排序两个环节因此全量数据写入性能相比原生桶写入约低 10%但在大量列中仅更新少量列的场景下列族表的写入效率远高于非列族表。注1.0 的部分更新功能也能通过只写变更列来规避这类写开销。读性能Read performance由于数据在写入时已排序读取时可直接使用 SortMerge 方式合并数据相比原生桶读取性能提升明显且内存消耗显著下降。Compaction 性能Compaction 与读取的逻辑一致跨列族的 Compaction 成本节约才是真正的大头收益。此外RFC-80 还提出 log 合并策略应当可插拔在哈希合并与排序合并之间可配置选择为此需要引入新的 log header 或标准机制来标记 base file / log file 是否已排序。约束与限制在设计上列族功能有以下明确边界RFC-80 原文整体设计依赖 Hudi 1.0 的非阻塞并发写入特性低版本 Hudi 无法读写列族表仅 MOR 桶表bucketed table支持设置列族。MOR Bucket 组合更合适因为其写性能更高但这并不意味着列族与其他索引及 COW 表完全不兼容列族不支持重分区repartitioning与重命名renamingSchema 演进Schema evolution不会自动作用于列族表——用户仍可显式地增删列与原生桶表一样列族表不支持 clustering 操作。存储格式变化文件命名规则引入列族后整个 Hudi 桶表的存储结构发生变化按 columnFamily 拆分 fileGroup 后base file 与 log file 的命名规则随之改变在所有文件名末尾追加_cfName后缀方便 Hudi 自身区分列族若没有该后缀则视为默认列族default column family。新的文件名模板为Base file[file_id]_[write_token]_[begin_time][_cfName].[extension]Log file[file_id]_[begin_instant_time][_cfName].log.[version]_[write_token]这与当前仓库中文件名的生成逻辑一脉相承。在 FSUtils.java 中可以看到现有的命名实现makeBaseFileName按%s_%s_%s%sfileId_writeToken_instantTime.extension拼接L119-L121makeLogFileName按%s_%s%s.%d_%sfileId_deltaCommitTime.log.version_writeToken拼接L423-L428日志文件匹配的正则LOG_FILE_PATTERN形如^\.(.)_(.*)\.(log|archive)\.(\d)...L81-L82。从源码结构可以推断列族落地时只需在上述模板的文件名末尾追加_cfName并同步演进元数据表metadata table的文件 schema额外记录一列 column family 名称即可。创建表时指定列族在表创建语句中通过 options / tblproperties 属性以key-value模式指定列族划分Key为列族名格式为hoodie.colFamily.名称列族命名规则按文档约定Value为列族具体内容由该列族包含的全部列 preCombine 字段组成格式为col1,col2...colN; precombineCol——列族列清单与 preCombine 字段用;分隔列清单内部用,分隔。约束条件列族列清单必须包含主键primary key除主键外不同列族包含的列不能重叠preCombine 字段可以不指定缺省时默认取主键。建表完成后列族属性会被持久化到 Hudi 的元数据中供后续读写使用。已有表中添加 / 删除列族对已存在的表使用 SQL 的 ALTER 命令修改列族属性并持久化新增列族ALTER TABLE table_name SET TBLPROPERTIES (hoodie.columnFamily.ka,b,c;a);删除列族ALTER TABLE table_name UNSET TBLPROPERTIES(hoodie.columnFamily.k);具体执行步骤如下执行 ALTER 命令修改列族校验 ALTER 后的列族是否合法必须满足以下条件否则校验不通过已有列族的名称不能修改其他列族中的列不能被划入新的列族新建列族必须满足上文介绍的格式要求包含主键、不重叠等将修改后的列族保存到.hoodie目录中。数据写入排序、拆分与列族级追加Hudi 内核按列族划分输入数据属于某个列族的数据经过排序后直接写入对应列族的 log 文件。具体写入步骤分桶与 Shuffle引擎按哈希将待写数据划分为桶并做 shuffle由写入引擎自行完成与原生桶写入一致排序Hudi 内核按主键对写入每个桶的数据排序Spark 与 Flink 均有各自的ExternalSorter可直接参考实现切分排序完成后将数据拆分到各列族写入将切分后的数据写入对应列族的 log 文件。这一流程在现有代码中已有对应底座MOR 表的追加写入由 HoodieAppendHandle.java 承载doAppend、canWrite、write等核心方法见 L438-L558桶写入侧则有BucketBulkInsertDataInternalWriterHelper等专用 Helper 类支撑见 hudi-spark-client 下的 BucketBulkInsertDataInternalWriterHelper.java其中通过BucketIdentifier.getBucketId(...)完成哈希分桶。通用 API 接口ColumnFamilyAppendHandle表列被聚簇后写入流程相比原生桶写入增加了排序与拆分环节因此需要引入新的追加接口来支持列族。RFC-80 提出的设计是引入ColumnFamilyAppendHandle继承AppendHandle实现列族写入。数据读取ColumnFamilyReader 与 RowReader 两级合并Hudi 内核的行读取row reader分为以下步骤Hudi 按列族组织待读文件引入familyReader合并读取每个列族自己的 baseFile 与 logFile实现列族级数据读取由于 log 文件按主键排序写入familyReader通过主键sortMerge合并自己的 baseFile 与 logFilefamilyReader支持上游传入的列裁剪column pruning减少读取时的 IO 开销合并过程中若用户为列族指定了 preCombine 字段则基于该字段选择合并策略这一逻辑复用 Hudi 自身已有的 precombine 逻辑无需修改RowReader按主键合并多个familyReader读取到的数据。由于每个familyReader读出的数据已按主键有序RowReader采用sortMergeJoin形式合并各familyReader的数据并返回完整行。整个读取过程涉及大量数据合并但得益于数据本身有序合并过程的内存消耗极低、速度很快相比 Hudi 原生合并方式内存压力与合并耗时显著降低。引擎读取的伪流程引擎将需要扫描的数据文件分发给 executor / worker / taskmanagerexecutor/worker/taskmanager 调用 Hudi 的RowReader接口并传入列裁剪与过滤条件Hudi 内核完成RowReader的数据读取并返回完整数据数据格式为Avro引擎拿到 Avro 数据后转换为自身所需格式Spark 转UnsafeRow、hetu 转block、Flink 转row、Hive 转ArrayWritable。关于 precombine 部分更新的语义仓库中有现成实现可佐证PartialUpdateAvroPayload.java 定义了preCombine/combineAndGetUpdateValue/overwriteField等行为并在对应测试 TestPartialUpdateAvroPayload.java 与 Spark 侧 TestPartialUpdateAvroPayload.scala 中得到验证。列族级 Compaction 与全量 Compaction列族级 CompactionFamily Compaction扩展 Hudi 的 compaction 调度模块合并每个列族自己的 base file 与 log file全量 CompactionGrand Compaction扩展 Hudi 的 compaction 调度模块合并更新整张表的所有列族。列族级合并完成后将多个列族最终合并为一行完整数据并保存。关键设计取舍Full compaction 是可选项只有列族级 compaction 是必需的不同用户需要的列可能不同有人需要来自多个列族的列有人只需要单个列族的列因此更合理的做法是允许用户自行选择是否开启 full compaction另外full compaction 之后读者侧的投影projection效率会下降因为此时投影只能基于包含全部字段的完整 parquet 文件进行而无法再按列族名做裁剪应用场景决定取舍列族不仅可用于 ML 场景和 AI 特征表也可用于模拟多流 Join 与宽表拼接。在模拟多流 Join 场景下Hudi 需要产出完整行因此这种情况下必须开启 full compaction。从现有代码结构看compaction 相关实现集中在 hudi-client-common 的 table/action/compact 目录包含ScheduleCompactionActionExecutor调度、RunCompactionActionExecutor执行、HoodieCompactor、CompactionTriggerStrategy等模块RFC-80 中扩展 compaction 调度模块的设计即落在这些类之上。兼容性与采纳计划列族是一个全新特性默认不开启业务行为兼容无影响。功能不会被主动开启未定义列族时不会进入列族逻辑路径语法兼容无影响。列族属性位于表属性table attributes中通过 SQL 标准语法执行数据类型处理方式兼容无影响。该设计不会修改底层数据字段格式。从配置角度多 writer 并发写入的能力由hoodie.write.concurrency.mode控制其定义见 HoodieWriteConfig.java 的 WRITE_CONCURRENCY_MODEL580-L583默认值为SINGLE_WRITER可选OPTIMISTIC_CONCURRENCY_CONTROL与NON_BLOCKING_CONCURRENCY_CONTROL。由于列族依赖 1.0 的非阻塞并发写入使用列族时需将写并发模式配置为相应取值相关行为测试可参考 TestSparkNonBlockingConcurrencyControl.java。测试计划与验收清单RFC-80 给出了实现验收的检查清单可作为功能自测参考所有未定义列族的表上现有测试全部通过Hudi SQL 支持在创建 MOR 桶表时设置列族MOR 桶表支持通过 SQL 添加 / 删除列族Hudi 支持按列族写入数据Hudi 支持按列族读取数据Hudi 支持列族级 compactionHudi 支持全量 compaction合并所有列族的数据实现数据加宽data widening。总结RFC-80 为 Hudi 引入了列族这一存储组织维度在桶表 fileGroup 内部按列聚簇拆分文件通过_cfName后缀标识列族并在写入时按主键排序、按列族切分落盘读取时借助 familyReader RowReader 两级 sortMerge 完成高效合并。它面向实时宽表拼接、AI 特征表、多流 Join 模拟等场景在保留 MOR 桶表高写入性能的同时大幅降低 Compaction 写放大与读取合并的内存压力。对于需要按列裁剪读取、并希望控制状态后端压力的流式宽表场景列族机制是一个值得跟进并实战验证的技术方向。赞分享大数据数据湖数据工程【免费下载链接】hudiUpserts, Deletes And Incremental Processing on Big Data.项目地址https://gitcode.com/gh_mirrors/hu/hudi点击查看免费下载相关推荐Hudi RFC-48 解读Log Compaction 支持与 MOR 表写放大优化Hudi RFC 48 解读Log Compaction 支持与 MOR 表写放大优化 本文以 Apache Hudi 官方设计文档 rfc 48.md ht数据湖湖仓一体大数据数据存储掌握列宽TanStack Vue Table 的 Column Sizing 特性完整实战指南掌握列宽TanStack Vue Table 的 Column Sizing 特性完整实战指南 Column Sizing列宽是 TanStack Vue前端UI组件Svelte 表格列宽度Column Sizing实战指南基于 tanstack/svelte-table 的列宽测量与状态管理Svelte 表格列宽度Column Sizing实战指南基于 tanstack/svelte table 的列宽测量与状态管理 TanStack Ta前端UI组件上一篇QEMU 持久化预留管理器Persistent Reservation Manager实战指南scsi-block/scsi-generic 与 qemu-pr-helper 协作机制下一篇如何5分钟快速上手Switch游戏文件管理工具NSC_BUILDER完整指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
