Article
第1章：Hudi 概述与核心概念
1.1 什么是 Hudi？
| 概念名称 | 说明 | 注意事项 |
|---|---|---|
| Apache Hudi | Hudi（Hadoop Upserts Deletes and Incrementals）是一个开源的数据管理框架，用于在数据湖上实现高效的记录级插入、更新、删除和增量查询。它运行在 HDFS 或云存储之上，与 Spark、Flink 等计算引擎集成。 | Hudi 不是一个数据库，而是一个构建在数据湖之上的”表格式”（Table Format），类似 Delta Lake 或 Iceberg。 |
| 开源项目 | 由 Uber 开发并开源，后捐赠给 Apache 软件基金会，目前为 Apache 顶级项目。 | 社区活跃，版本迭代较快，建议关注官方文档与发布日志。 |
| 核心目标 | 解决传统数据湖无法高效支持”更新”和”删除”的问题，提供近实时的数据写入与查询能力。 | 适用于需要”准实时”或”微批”处理的场景，而非纯流式处理。 |
1.2 Hudi 的核心价值与应用场景
| 概念名称 | 说明 | 注意事项 |
|---|---|---|
| 核心价值：高效更新/删除 | 支持记录级别的 upsert 和 delete，避免全表重写，显著提升数据新鲜度。 | 适用于用户行为日志、订单状态更新等需要修正历史数据的场景。 |
| 核心价值：增量处理 | 提供增量拉取接口，下游任务可仅消费自上次以来变更的数据，降低资源开销。 | 增量查询依赖 .hoodie 时间轴元数据，需确保其完整性。 |
| 核心价值：数据一致性 | 通过事务机制保证写入的原子性，避免部分写入导致的数据不一致。 | 需配置合适的并发控制策略以避免写冲突。 |
| 应用场景：实时数仓 | 构建湖仓一体架构中的实时分层表（如 DWD、DWS），支持快速数据修正。 | 适合与 Spark Structured Streaming 或 Flink CDC 集成。 |
| 应用场景：CDC 同步 | 将数据库的变更日志（Change Data Capture）同步到数据湖中，实现异构系统间的数据一致性。 | 需配合 Debezium、Flink CDC 等工具使用。 |
| 应用场景：数据修正 | 支持对已写入的数据进行”软删除”或”硬删除”，满足数据合规性要求（如 GDPR）。 | 删除操作需合理配置清理策略，避免小文件问题。 |
1.3 Hudi 与传统数据湖的对比
| 对比维度 | 传统数据湖（如 Parquet + Hive） | Hudi | 注意事项 |
|---|---|---|---|
| 更新支持 | 不支持记录级更新，需重写整个分区或文件 | 支持记录级 upsert 和 delete | Hudi 通过索引定位记录，避免全表扫描 |
| 删除支持 | 无法直接删除记录，需重写文件 | 支持标记删除和物理删除 | 删除性能优于传统方式 |
| 数据新鲜度 | 批处理延迟高（小时级） | 支持分钟级甚至秒级更新 | 依赖写入频率与 compaction 策略 |
| 增量处理 | 需依赖时间字段或分区字段模拟增量 | 原生支持基于时间轴的增量查询 | Hudi 增量查询更精确，避免漏读或重读 |
| 存储开销 | 写入简单，但更新成本高 | 引入 log 文件和索引，存储略高 | 可通过压缩与清理策略优化 |
| 查询性能 | 读取 Parquet 文件，性能稳定 | COPY_ON_WRITE 表接近原生性能，MERGE_ON_READ 表需合并 log 文件 | 查询性能受 compaction 策略影响 |
1.4 Hudi 支持的数据表类型（COPY_ON_WRITE、MERGE_ON_READ）
| 表类型 | 说明 | 注意事项 |
|---|---|---|
| COPY_ON_WRITE (COW) | 每次写入时，将新数据与旧文件中的数据合并，生成新的 base file。读取时无需合并操作，读性能高。 | 写放大较严重，适合写少读多场景。 |
| MERGE_ON_READ (MOR) | 新数据写入 log 文件，base file 不立即更新。读取时按需合并 base file 与 log file，支持两种查询模式。 | 写性能高，但读性能受合并影响，适合写多读少场景。 |
| 查询模式兼容性 | COW 表仅支持 Snapshot Query | MOR 表支持 Snapshot Query 和 Read Optimized Query |
1.5 Hudi 的核心组件（Timeline、File Groups、Index 等）
| 组件名称 | 说明 | 注意事项 |
|---|---|---|
| Timeline（时间轴） | 记录所有写入、压缩、清理等操作的元数据，是 Hudi 实现 ACID 和增量查询的核心。 | 所有操作必须通过时间轴协调，确保一致性。 |
| File Groups（文件组） | 每个文件组包含一个 base file 和多个 log file，对应一个数据分片。 | 文件组数量影响并行度与小文件问题。 |
| Base File | 存储当前最新版本的列式数据文件（如 Parquet）。 | COW 表每次写入生成新 base file；MOR 表定期通过 compaction 生成。 |
| Log File | 存储增量数据（如 upsert 记录），格式为 Avro。 | MOR 表特有，用于延迟合并。 |
| Index（索引） | 用于快速定位某条记录所在的文件组，避免全表扫描。 | 不同索引类型性能差异大，需根据场景选择。 |
| Hoodie Metadata | 可选组件，存储文件列表、统计信息等，加速元数据操作。 | 启用后可显著提升 listing 性能，但增加写入开销。 |
第2章：Hudi 架构与工作原理
2.1 Hudi 架构概览
| 组件名称 | 说明 | 注意事项 |
|---|---|---|
| 数据存储层 | HDFS 或云存储（S3、ABFS、GCS），存储 base file、log file 和 .hoodie 元数据目录。 | 需保证存储系统的高可用性。 |
| 计算引擎 | Spark、Flink、Hive 等，用于执行写入、查询、压缩等任务。 | Spark 是最成熟的集成方式。 |
| Hudi Client | 提供写入、读取、压缩、清理等核心逻辑，封装在 hudi-client 模块中。 | 客户端负责与时间轴交互并执行策略。 |
| .hoodie 目录 | 存储时间轴元数据、索引、文件组信息等，是 Hudi 表的”元数据中心”。 | 不可手动修改，否则可能导致表损坏。 |
| Timeline Server（可选） | 集中式服务，用于管理时间轴，支持高并发写入。 | 适用于大规模生产环境。 |
2.2 Hudi 时间轴（Timeline）机制
| 操作类型 | 说明 | 注意事项 |
|---|---|---|
| COMMIT | 表示一次写入操作（如 insert、upsert）成功提交。 | 每个 commit 有唯一 instant time（时间戳）。 |
| DELTA_COMMIT | 表示一次增量写入（主要针对 MOR 表的 log 写入）。 | 包含写入的 log 文件列表。 |
| COMPACTION | 表示一次 compaction 操作，将 log file 合并到 base file。 | 可异步执行，不影响写入。 |
| CLEAN | 表示一次清理操作，删除过期的文件版本。 | 清理策略可配置保留的 commits 数量。 |
| ROLLBACK | 表示一次回滚操作，撤销失败的写入。 | 用于恢复数据一致性。 |
| SAVEPOINT | 手动打的快照点，用于数据恢复。 | 可防止被清理策略删除。 |
| ARCHIVE | 表示一次归档操作，将旧的 timeline 元数据归档。 | 避免 .hoodie 目录过大。 |
2.3 Hudi 文件组织结构（Base Files、Log Files）
| 文件类型 | 说明 | 注意事项 |
|---|---|---|
| Base File (.parquet) | 存储主数据，列式存储格式，用于快速扫描查询。 | COW 表每次写入生成新 base file；MOR 表通过 compaction 生成。 |
| Log File (.log) | 存储增量变更记录，行式 Avro 格式，用于 MOR 表的延迟合并。 | 支持多种压缩格式（如 zlib、snappy、lz4）。 |
| .hoodie/<instant_time>.commit | 记录一次 commit 的元数据，包含写入的文件列表。 | 是增量查询的主要依据。 |
| .hoodie/<instant_time>.deltacommit | 记录一次 delta commit 的元数据。 | MOR 表特有。 |
| .hoodie/<instant_time>.clean | 记录一次清理操作删除的文件。 | 可用于审计。 |
| .hoodie/<instant_time>.rollback | 记录回滚操作涉及的文件。 | 用于故障恢复。 |
| .hoodie/archived_instants | 归档的时间轴记录，减少主目录压力。 | 不影响正常读写。 |
2.4 Hudi 索引机制（Indexing）
| 索引类型 | 说明 | 注意事项 |
|---|---|---|
| SIMPLE | 基于 record key 的哈希映射，存储在内存或外部系统中。 | 适合小数据量，简单高效。 |
| BLOOM | 使用布隆过滤器快速判断某 record key 是否在某个文件中。 | 默认索引类型，适合大多数场景。 |
| GLOBAL_BLOOM | 全局布隆过滤器，跨所有文件组检查 record key。 | 避免重复插入，但写入性能略低。 |
| HBASE | 将索引存储在外部 HBase 集群中，支持高并发写入。 | 适合超大规模表，需维护 HBase 集群。 |
| INMEMORY | 将索引全量加载到内存，查询最快。 | 内存消耗大，不适合大数据量。 |
| CUSTOM | 支持自定义索引实现。 | 需实现 HoodieIndex 接口。 |
2.5 Hudi 写操作流程（Insert、Upsert、Delete）
| 写操作 | 说明 | 注意事项 |
|---|---|---|
| INSERT | 将新数据写入，不检查是否已存在相同 record key。 | 写入性能最高，适用于无主键场景。 |
| UPSERT | 先通过索引查找 record key 是否存在，存在则更新，否则插入。 | 最常用操作，确保数据唯一性。 |
| UPDATE | 仅更新已存在的 record key，若不存在则报错。 | 使用较少，通常用 upsert 替代。 |
| DELETE | 标记某 record key 为删除状态（写入 delete marker 到 log file）。 | 物理删除需等待清理操作。 |
| BULK_INSERT | 批量插入大量数据，使用排序优化减少文件碎片。 | 适合首次加载历史数据，性能优于 insert。 |
| 写流程步骤 | 1. 构建索引 → 2. 定位文件组 → 3. 写入 base/log → 4. 提交 commit 到 timeline | 所有步骤需原子完成，否则触发 rollback |
第3章：Hudi 表类型与查询类型
3.1 COPY_ON_WRITE 表详解
| 方法/参数 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.table.type | COPY_ON_WRITE | 指定表类型为 COW，每次写入生成新 base file | .option(“hoodie.table.type”, “COPY_ON_WRITE”) | 适合读多写少场景，写放大明显 |
| hoodie.datasource.write.table.type | 同上 | Spark 写入时指定表类型 | .option(“hoodie.datasource.write.table.type”, “COPY_ON_WRITE”) | 建表时必须显式设置 |
| hoodie.compact.inline | true / false | 是否开启同步压缩（对 COW 无效） | .option(“hoodie.compact.inline”, “false”) | COW 表无需压缩，此参数可关闭 |
| hoodie.parquet.small.file.size | 字节数（如 134217728） | 定义小文件阈值，触发合并 | .option(“hoodie.parquet.small.file.size”, “134217728”) | 避免产生过多小文件 |
| hoodie.parquet.max.file.size | 字节数（如 536870912） | 单个 Parquet 文件最大大小 | .option(“hoodie.parquet.max.file.size”, “536870912”) | 影响并行读取性能 |
3.2 MERGE_ON_READ 表详解
| 方法/参数 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.table.type | MERGE_ON_READ | 指定表类型为 MOR | .option(“hoodie.table.type”, “MERGE_ON_READ”) | 适合写多读少场景 |
| hoodie.datasource.write.table.type | 同上 | Spark 写入时指定表类型 | .option(“hoodie.datasource.write.table.type”, “MERGE_ON_READ”) | 必须与 hoodie.table.type 一致 |
| hoodie.compact.inline | true | 是否开启同步压缩 | .option(“hoodie.compact.inline”, “true”) | 建议开启，避免 log 文件过多 |
| hoodie.compact.inline.max.delta.commits | 整数（如 5） | 每 N 次 delta commit 触发一次压缩 | .option(“hoodie.compact.inline.max.delta.commits”, “5”) | 控制压缩频率 |
| hoodie.log.format.protobuf.enable | true / false | 是否使用 Protobuf 格式存储 log 文件 | .option(“hoodie.log.format.protobuf.enable”, “true”) | 可提升 log 写入效率 |
| hoodie.log.block.size | 字节数（如 104857600） | log 文件 block 大小 | .option(“hoodie.log.block.size”, “104857600”) | 影响 log 合并性能 |
3.3 查询类型：Snapshot Query
| 方法/参数 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.datasource.query.type | snapshot | 启用快照查询，读取最新合并数据 | .option(“hoodie.datasource.query.type”, “snapshot”) | 默认查询类型 |
| DataSourceReadOptions.QUERY_TYPE_SNAPSHOT_OPT_VAL() | Scala API 常量 | 等价于 “snapshot” | .option(DataSourceReadOptions.QUERY_TYPE_KEY(), DataSourceReadOptions.QUERY_TYPE_SNAPSHOT_OPT_VAL()) | 代码中推荐使用常量 |
| 支持的表类型 | COW 和 MOR | COW 直接读 base file；MOR 自动合并 log file | spark.read.format(“hudi”).load(“/path”) | MOR 表合并可能影响延迟 |
| 读取方式 | Spark DataFrame 或 Spark SQL | 通过 path 或 Hive 表名读取 | spark.sql(“SELECT * FROM hudi_table”) | 需确保 Hive 同步已完成 |
3.4 查询类型：Incremental Query
| 方法/参数 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.datasource.query.type | incremental | 启用增量查询 | .option(“hoodie.datasource.query.type”, “incremental”) | 仅支持 MOR 和 COW 表 |
| hoodie.datasource.read.begin.instanttime | 时间戳（如 20250101000000） | 指定起始 commit 时间 | .option(“hoodie.datasource.read.begin.instanttime”, “20250101000000”) | 不包含该时间点 |
| hoodie.datasource.read.end.instanttime | 时间戳 | 指定结束 commit 时间（可选） | .option(“hoodie.datasource.read.end.instanttime”, “20250101010000”) | 包含该时间点 |
| hoodie.inc.removed.rownames | true/false | 是否返回被删除的 record key | .option(“hoodie.inc.removed.rownames”, “true”) | 用于同步删除操作 |
| 用途 | 获取自某时间点以来的所有变更 | 用于下游 ETL 增量处理 | spark.read.format(“hudi”).option(“hoodie.datasource.query.type”, “incremental”).option(“hoodie.datasource.read.begin.instanttime”, “20250101000000”).load(basePath) | begin instant time 必须存在且未被归档 |
3.5 查询类型：Read Optimized Query
| 方法/参数 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.datasource.query.type | read_optimized | 启用读优化查询，仅读取 base file | .option(“hoodie.datasource.query.type”, “read_optimized”) | 仅适用于 MOR 表 |
| DataSourceReadOptions.QUERY_TYPE_READ_OPTIMIZED_OPT_VAL() | Scala 常量 | 等价于 “read_optimized” | .option(DataSourceReadOptions.QUERY_TYPE_KEY(), DataSourceReadOptions.QUERY_TYPE_READ_OPTIMIZED_OPT_VAL()) | 推荐使用常量 |
| 数据延迟 | 高 | 不读取 log file，数据可能过时 | spark.read.format(“hudi”).option(“hoodie.datasource.query.type”, “read_optimized”).load(“/path”) | 适用于对延迟不敏感的报表场景 |
| 性能 | 高 | 无需合并 log，读取速度快 | 适合大规模扫描查询 | |
| 与 Snapshot 对比 | 数据旧但快 | Snapshot 数据新但需合并 | 根据业务需求选择 |
第4章：Hudi 写操作 API（Spark DataSource）
4.1 使用 Spark 写入 Hudi 表（基本语法）
| 方法/参数 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| format(“hudi”) | spark.write.format(“hudi”) | 指定数据源为 Hudi | df.write.format(“hudi”).option(“hoodie.table.name”, “my_table”).mode(SaveMode.Append).save(“/path”) | 必须使用 hudi 数据源 |
| mode(SaveMode.Append) | Append / Overwrite | 写入模式 | .mode(SaveMode.Append) | Append 为常用模式，避免覆盖元数据 |
| hoodie.table.name | 表名字符串 | 指定 Hudi 表名 | .option(“hoodie.table.name”, “sales”) | 必须设置 |
| hoodie.datasource.write.table.name | 同上 | 替代写法 | .option(“hoodie.datasource.write.table.name”, “sales”) | 与 hoodie.table.name 等效 |
| path | 存储路径 | 指定 Hudi 表在存储系统中的路径 | .save(“/user/hive/warehouse/sales”) | 路径下会生成 .hoodie 目录 |
4.2 插入数据（INSERT）
| 方法/参数 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.datasource.write.operation | insert | 执行插入操作，不查重 | .option(“hoodie.datasource.write.operation”, “insert”) | 性能最高 |
| hoodie.insert.shuffle.parallelism | 并行度数值 | 设置插入 shuffle 并行度 | .option(“hoodie.insert.shuffle.parallelism”, “10”) | 建议等于目标文件数 |
| 适用场景 | 无主键或允许重复数据 | 快速加载原始日志 | df.write.format(“hudi”).option(“hoodie.datasource.write.operation”, “insert”).save(path) | 不保证 record key 唯一性 |
4.3 插入更新（UPSERT）
| 方法/参数 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.datasource.write.operation | upsert | 插入或更新，基于 record key | .option(“hoodie.datasource.write.operation”, “upsert”) | 最常用写操作 |
| hoodie.upsert.shuffle.parallelism | 并行度数值 | 设置 upsert shuffle 并行度 | .option(“hoodie.upsert.shuffle.parallelism”, “10”) | 影响写性能 |
| hoodie.datasource.write.payload.class | Payload 类名 | 指定如何合并新旧记录 | .option(“hoodie.datasource.write.payload.class”, “org.apache.hudi.common.model.DefaultHoodieRecordPayload”) | 默认策略为覆盖 |
| hoodie.combine.before.upsert | true/false | 是否在 upsert 前合并输入数据 | .option(“hoodie.combine.before.upsert”, “true”) | 减少索引查询次数，提升性能 |
4.4 更新数据（UPDATE）
| 方法/参数 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.datasource.write.operation | update | 仅更新已存在的 record key | .option(“hoodie.datasource.write.operation”, “update”) | 若 record key 不存在则失败 |
| hoodie.update.shuffle.parallelism | 并行度数值 | 设置更新操作的并行度 | .option(“hoodie.update.shuffle.parallelism”, “10”) | 需根据数据量调整 |
| 使用建议 | 场景较少 | 通常用 upsert 替代 | 避免使用，除非明确要求 |
4.5 删除数据（DELETE）
| 方法/参数 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.datasource.write.operation | delete | 执行删除操作 | .option(“hoodie.datasource.write.operation”, “delete”) | 仅标记删除 |
| hoodie.datasource.write.payload.class | org.apache.hudi.EmptyHoodieRecordPayload | 使用空 payload 实现删除 | .option(“hoodie.datasource.write.payload.class”, “org.apache.hudi.EmptyHoodieRecordPayload”) | 必须配合 delete 操作 |
| hoodie.delete.shuffle.parallelism | 并行度数值 | 设置删除操作的并行度 | .option(“hoodie.delete.shuffle.parallelism”, “10”) | 影响删除性能 |
| 物理删除 | 通过 clean 操作 | 清理过期文件实现物理删除 | 删除非即时生效 |
4.6 批量插入（BULK_INSERT）
| 方法/参数 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.datasource.write.operation | bulk_insert | 高效批量插入大量数据 | .option(“hoodie.datasource.write.operation”, “bulk_insert”) | 适合首次加载 |
| hoodie.bulkinsert.shuffle.parallelism | 并行度数值 | 设置 bulk insert 并行度 | .option(“hoodie.bulkinsert.shuffle.parallelism”, “20”) | 建议设为文件目标数的倍数 |
| hoodie.datasource.write.row.writer.enable | true/false | 是否启用行级写入器（提升性能） | .option(“hoodie.datasource.write.row.writer.enable”, “true”) | 推荐开启 |
| hoodie.bulkinsert.user.defined.partitioner.class | 分区类名 | 自定义分区策略 | .option(“hoodie.bulkinsert.user.defined.partitioner.class”, “com.example.MyPartitioner”) | 用于控制数据分布 |
| 优势 | 无索引查找，性能极高 | 适合 TB 级历史数据导入 | 不支持更新语义，仅用于初始加载 |
第5章：Hudi 配置参数详解
5.1 基础配置（表名、路径、键字段等）
| 参数名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.table.name | 字符串 | 指定 Hudi 表的逻辑名称 | .option(“hoodie.table.name”, “users”) | 必须设置，用于元数据记录 |
| hoodie.datasource.write.table.name | 字符串 | 同上，Spark 写入时使用 | .option(“hoodie.datasource.write.table.name”, “users”) | 与 hoodie.table.name 等效 |
| path | 存储路径 | 指定表在文件系统中的根路径 | .save(“/data/hudi/users”) | 路径下自动生成 .hoodie 目录 |
| hoodie.keygenerator.class | 类名 | 指定 key 生成器类 | .option(“hoodie.keygenerator.class”, “org.apache.hudi.keygen.ComplexKeyGenerator”) | 默认为 SimpleKeyGenerator |
| hoodie.datasource.write.recordkey.field | 字段名（逗号分隔） | 指定 record key 字段 | .option(“hoodie.datasource.write.recordkey.field”, “user_id”) | 唯一标识一条记录 |
| hoodie.datasource.write.partitionpath.field | 字段名（逗号分隔） | 指定分区字段 | .option(“hoodie.datasource.write.partitionpath.field”, “dt,region”) | 支持多级分区 |
| hoodie.datasource.write.hive_style_partitioning | true/false | 是否启用 Hive 风格分区（dt=2025-01-01） | .option(“hoodie.datasource.write.hive_style_partitioning”, “true”) | 推荐开启以兼容 Hive |
5.2 并行度与性能调优参数
| 参数名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.insert.shuffle.parallelism | 整数 | 设置 insert 操作的 shuffle 并行度 | .option(“hoodie.insert.shuffle.parallelism”, “10”) | 建议等于目标文件数 |
| hoodie.upsert.shuffle.parallelism | 整数 | 设置 upsert 操作的并行度 | .option(“hoodie.upsert.shuffle.parallelism”, “20”) | 影响写入性能 |
| hoodie.delete.shuffle.parallelism | 整数 | 设置 delete 操作的并行度 | .option(“hoodie.delete.shuffle.parallelism”, “10”) | 根据删除数据量调整 |
| hoodie.bulkinsert.shuffle.parallelism | 整数 | 设置 bulk_insert 并行度 | .option(“hoodie.bulkinsert.shuffle.parallelism”, “30”) | 建议为文件目标数的倍数 |
| spark.sql.adaptive.enabled | true | 启用 Spark AQE（自适应查询执行） | spark.conf.set(“spark.sql.adaptive.enabled”, true) | 提升读写性能 |
| spark.serializer | org.apache.spark.serializer.KryoSerializer | 使用 Kryo 序列化提升性能 | spark.conf.set(“spark.serializer”, “org.apache.spark.serializer.KryoSerializer”) | 推荐在 Spark 作业中启用 |
5.3 压缩与清理策略
| 参数名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.compact.inline | true/false | 是否开启同步压缩（写入时触发） | .option(“hoodie.compact.inline”, “true”) | MOR 表建议开启 |
| hoodie.compact.inline.max.delta.commits | 整数 | 每 N 次 delta commit 触发一次压缩 | .option(“hoodie.compact.inline.max.delta.commits”, “5”) | 控制压缩频率 |
| hoodie.cleaner.policy | KEEP_LATEST_COMMITS / KEEP_LATEST_FILE_VERSIONS | 清理策略类型 | .option(“hoodie.cleaner.policy”, “KEEP_LATEST_COMMITS”) | 默认为保留最新 commits |
| hoodie.cleaner.commits.retained | 整数 | 保留最近 N 次 commit | .option(“hoodie.cleaner.commits.retained”, “10”) | 避免增量查询断档 |
| hoodie.cleaner.fileversions.retained | 整数 | 保留每个文件的最近 N 个版本 | .option(“hoodie.cleaner.fileversions.retained”, “3”) | 防止误删 |
| hoodie.clean.async | true/false | 是否异步清理 | .option(“hoodie.clean.async”, “true”) | 减少写入阻塞 |
5.4 索引配置（Simple、Bloom、Global 等）
| 参数名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.index.type | BLOOM / GLOBAL_BLOOM / SIMPLE / HBASE | 指定索引类型 | .option(“hoodie.index.type”, “GLOBAL_BLOOM”) | 默认为 BLOOM |
| hoodie.bloom.index.parallelism | 整数 | Bloom 索引构建并行度 | .option(“hoodie.bloom.index.parallelism”, “2”) | 小表可设为 1 |
| hoodie.index.key.fields | 字段名（逗号分隔） | 指定用于索引的 key 字段 | .option(“hoodie.index.key.fields”, “user_id”) | 通常与 record key 一致 |
| hoodie.bloom.filter.type | SIMPLE / DYNAMIC | 布隆过滤器类型 | .option(“hoodie.bloom.filter.type”, “DYNAMIC”) | DYNAMIC 更节省空间 |
| hoodie.index.global.enabled | true/false | 是否启用全局索引（跨分区） | .option(“hoodie.index.global.enabled”, “true”) | 允许跨分区去重 |
| hoodie.hbase.index.zkquorum | ZooKeeper 地址 | HBase 索引使用的 ZooKeeper 集群 | .option(“hoodie.hbase.index.zkquorum”, “zk1:2181,zk2:2181”) | 仅 HBASE 索引需要 |
5.5 分区配置与分区策略
| 参数名称 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.datasource.write.partitionpath.field | 字段名（逗号分隔） | 指定分区字段 | .option(“hoodie.datasource.write.partitionpath.field”, “dt”) | 支持多级分区如 dt,region |
| hoodie.datasource.write.keygenerator.class | KeyGenerator 类名 | 指定 key 生成器 | .option(“hoodie.datasource.write.keygenerator.class”, “org.apache.hudi.keygen.TimestampKeyGenerator”) | 控制分区逻辑 |
| hoodie.datasource.write.hive_style_partitioning | true/false | 是否使用 Hive 风格分区路径 | .option(“hoodie.datasource.write.hive_style_partitioning”, “true”) | 路径格式为 dt=2025-01-01/ |
| hoodie.datasource.write.partition.fields | 字段名 | 同 partitionpath.field | .option(“hoodie.datasource.write.partition.fields”, “dt”) | 别名，功能相同 |
| hoodie.datasource.hive_sync.partition_fields | 字段名 | Hive 同步时使用的分区字段 | .option(“hoodie.datasource.hive_sync.partition_fields”, “dt”) | 确保 Hive 表结构正确 |
| 自定义分区器 | 实现 Partitioner 接口 | 自定义数据到分区的映射逻辑 | .option(“hoodie.datasource.write.partitionizer.class”, “com.example.CustomPartitioner”) | 高级用法，需编程实现 |
第6章：Hudi 查询操作
6.1 使用 Spark SQL 查询 Hudi 表
| 方法/参数 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| CREATE TABLE … USING hudi | DDL 语句 | 在 Hive Metastore 中注册 Hudi 表 | CREATE TABLE hudi_table USING hudi LOCATION ‘/path’; | 需启用 Hive 同步 |
| spark.sql(“SELECT * FROM table”) | SQL 查询 | 通过 Spark SQL 查询 Hudi 表 | spark.sql(“SELECT user_id, name FROM users WHERE dt=‘2025-01-01’“) | 支持标准 SQL 语法 |
| HoodieSparkSessionHelper | Scala/Java API | 在代码中注册临时视图 | spark.read.format(“hudi”).load(path).createOrReplaceTempView(“users_view”) | 用于 SQL 查询临时表 |
| spark.sql.hive.convertMetastoreParquet | true/false | 是否启用 Hive Parquet 转换 | spark.conf.set(“spark.sql.hive.convertMetastoreParquet”, false) | 避免与 Hudi 冲突 |
| 查询路径 | 直接 load 路径 | 不依赖 Hive 时直接读取路径 | spark.read.format(“hudi”).load(“/data/hudi/users”) | 推荐用于临时分析 |
6.2 使用 Spark DataFrame 查询 Snapshot
| 方法/参数 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.datasource.query.type=snapshot | 查询类型设置 | 读取最新快照数据 | spark.read.format(“hudi”).option(“hoodie.datasource.query.type”, “snapshot”).load(“/path”) | 默认行为 |
| DataSourceReadOptions.QUERY_TYPE_SNAPSHOT_OPT_VAL() | Scala 常量 | 获取 snapshot 查询类型值 | .option(DataSourceReadOptions.QUERY_TYPE_KEY(), DataSourceReadOptions.QUERY_TYPE_SNAPSHOT_OPT_VAL()) | 代码中更安全 |
| 读取方式 | DataFrame API | 返回标准 DataFrame | val df = spark.read.format(“hudi”).load(path) | 可进行 filter、agg 等操作 |
| 性能 | 高（COW）或中（MOR） | COW 直接读 base file；MOR 需合并 log | MOR 表合并可能影响延迟 | |
| 适用场景 | 实时报表、数据校验 | 获取当前最新状态 | 是最常用的查询方式 |
6.3 增量查询（Incremental Query）配置与使用
| 方法/参数 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.datasource.query.type=incremental | 启用增量查询 | 仅读取自某时间点以来的变更 | .option(“hoodie.datasource.query.type”, “incremental”) | 仅支持 COW 和 MOR 表 |
| hoodie.datasource.read.begin.instanttime | 时间戳字符串 | 指定起始 commit 时间 | .option(“hoodie.datasource.read.begin.instanttime”, “20250101000000”) | 必须存在且未归档 |
| hoodie.datasource.read.end.instanttime | 时间戳字符串 | 指定结束时间（可选） | .option(“hoodie.datasource.read.end.instanttime”, “20250101010000”) | 包含该时间点 |
| hoodie.inc.removed.rownames | true/false | 是否返回被删除的 record key | .option(“hoodie.inc.removed.rownames”, “true”) | 用于同步删除操作 |
| 获取 commit 时间 | 读取 .hoodie 目录 | 从文件系统获取可用的 instants | fs.listStatus(new Path(basePath + “/.hoodie”)) | 用于动态设置 begin time |
| 示例代码 | 完整调用 | 执行增量查询 | spark.read.format(“hudi”).option(“hoodie.datasource.query.type”, “incremental”).option(“hoodie.datasource.read.begin.instanttime”, “20250101000000”).load(basePath) | 可用于 CDC 下游消费 |
6.4 读优化查询（Read Optimized）使用场景
| 方法/参数 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.datasource.query.type=read_optimized | 启用读优化查询 | 仅读取 base file，不合并 log | .option(“hoodie.datasource.query.type”, “read_optimized”) | 仅适用于 MOR 表 |
| DataSourceReadOptions.QUERY_TYPE_READ_OPTIMIZED_OPT_VAL() | Scala 常量 | 获取 RO 查询类型值 | .option(DataSourceReadOptions.QUERY_TYPE_KEY(), DataSourceReadOptions.QUERY_TYPE_READ_OPTIMIZED_OPT_VAL()) | 推荐使用 |
| 数据延迟 | 高 | 不读取 log 文件，数据可能滞后 | spark.read.format(“hudi”).option(“hoodie.datasource.query.type”, “read_optimized”).load(path) | 适用于对实时性要求不高的场景 |
| 性能 | 高 | 无需合并操作，读取速度快 | 适合大规模扫描 | |
| 适用场景 | 离线报表、BI 分析 | 数据已通过 compaction 合并 | 需确保 compaction 策略合理 |
6.5 查询性能优化建议
| 优化策略 | 说明 | 配置建议 | 注意事项 |
|---|---|---|---|
| 启用 Metadata Table | 使用内置元数据表加速文件列表操作 | .option(“hoodie.metadata.enable”, “true”) | 显著提升 listing 性能 |
| 合理设置文件大小 | 避免小文件或超大文件 | hoodie.parquet.max.file.size=512MB | 平衡读取并行度与调度开销 |
| 使用布隆过滤器 | 减少无效文件扫描 | hoodie.bloom.index.filter=true | 默认开启，建议保留 |
| 分区剪枝 | 仅扫描相关分区 | 在 SQL 中使用分区字段过滤 | 如 WHERE dt=‘2025-01-01’ |
| 统计信息收集 | 启用列统计用于谓词下推 | Hudi 自动写入 Parquet 统计 | 确保查询可利用 |
| Spark 读取并行度 | 调整分区数量匹配文件数 | 使用 coalesce 或 repartition | 避免过多小任务 |
| 使用 COW 表 | 读性能最优 | 选择 COW 表类型 | 适用于写少读多场景 |
第7章：Hudi 数据管理与运维
7.1 数据清理（Clean）机制
| 参数/方法 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.cleaner.policy | KEEP_LATEST_COMMITS / KEEP_LATEST_FILE_VERSIONS | 指定清理策略 | .option(“hoodie.cleaner.policy”, “KEEP_LATEST_COMMITS”) | 推荐使用 COMMITS 策略 |
| hoodie.cleaner.commits.retained | 整数（如 10） | 保留最近 N 次 commit | .option(“hoodie.cleaner.commits.retained”, “10”) | 必须大于增量查询所需范围 |
| hoodie.cleaner.fileversions.retained | 整数（如 3） | 保留每个文件的最近 N 个版本 | .option(“hoodie.cleaner.fileversions.retained”, “3”) | 防止误删历史版本 |
| hoodie.clean.async | true/false | 是否异步执行清理 | .option(“hoodie.clean.async”, “true”) | 减少写入阻塞，推荐开启 |
| hoodie.cleaner.parallelism | 整数 | 清理操作的并行度 | .option(“hoodie.cleaner.parallelism”, “4”) | 根据集群资源调整 |
| 清理触发方式 | 自动或手动 | 写入时自动触发或单独执行 | CleanerTool 命令行工具 | 可通过 hudi-cli 手动执行 |
7.2 压缩（Compaction）机制
| 参数/方法 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.compact.inline | true/false | 是否同步执行压缩 | .option(“hoodie.compact.inline”, “true”) | MOR 表建议开启 |
| hoodie.compact.inline.max.delta.commits | 整数（如 5） | 每 N 次 delta commit 触发一次压缩 | .option(“hoodie.compact.inline.max.delta.commits”, “5”) | 控制压缩频率，避免过频 |
| hoodie.compaction.trigger.strategy | num_and_time / num_commits / time_elapsed | 压缩触发策略 | .option(“hoodie.compaction.trigger.strategy”, “num_and_time”) | num_and_time 更灵活 |
| hoodie.compaction.delta_commits | 整数 | 基于 commit 次数触发压缩 | .option(“hoodie.compaction.delta_commits”, “4”) | 配合 num_commits 策略使用 |
| hoodie.compaction.delta_seconds | 秒数（如 3600） | 基于时间间隔触发压缩 | .option(“hoodie.compaction.delta_seconds”, “3600”) | 防止 log 文件过大 |
| hoodie.compaction.parallelism | 整数 | 压缩任务的并行度 | .option(“hoodie.compaction.parallelism”, “10”) | 建议等于目标文件组数量 |
| 手动压缩 | 使用 CompactorUtil 或 CLI | 强制执行压缩 | spark-submit —class org.apache.hudi.utilities.HoodieCompactor … | 用于紧急优化查询性能 |
7.3 时间轴归档（Archiving）
| 参数/方法 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.archive.merge.enable | true/false | 是否启用归档合并 | .option(“hoodie.archive.merge.enable”, “true”) | 减少归档文件数量 |
| hoodie.archival.retired.delta.commits | 整数（如 20） | 保留最近 N 次 delta commits 后开始归档 | .option(“hoodie.archival.retired.delta.commits”, “20”) | 必须大于 cleaner.commits.retained |
| hoodie.archival.retired.commits.archival.batch | 整数（如 10） | 每次归档的 commit 批量数 | .option(“hoodie.archival.retired.commits.archival.batch”, “10”) | 控制归档操作粒度 |
| 归档路径 | .hoodie/.archive/ | 归档文件存储目录 | 自动创建于表根目录下 | 不可手动删除 |
| 影响 | 避免 .hoodie 目录过大 | 提升元数据操作性能 | 归档后仍可通过 API 访问历史元数据 | |
| 查询兼容性 | 支持 | 增量查询仍可跨归档段读取 | 设置 begin.instanttime 为归档时间点 | 需确保未被清理 |
7.4 数据回滚（Rollback）
| 参数/方法 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.rollback.using.spark | true/false | 是否使用 Spark 执行回滚 | .option(“hoodie.rollback.using.spark”, “true”) | 大表建议开启 |
| hoodie.rollback.parallelism | 整数 | 回滚操作的并行度 | .option(“hoodie.rollback.parallelism”, “10”) | 影响回滚速度 |
| 触发条件 | 写入失败或手动触发 | 撤销上一次失败的 commit | 自动触发或通过 hudi-cli 手动执行 | 回滚会删除写入的文件 |
| 回滚内容 | 文件与元数据 | 删除失败写入的文件并更新 timeline | .hoodie/<failed_instant>.rollback | 保证原子性 |
| 限制 | 仅对未完成或失败的 commit 有效 | 成功提交的 commit 无法直接回滚 | 需结合 clean 和 restore 实现 | 回滚不是”撤销”操作 |
| 监控 | 查看 .rollback 文件 | 确认回滚是否成功 | fs.exists(new Path(rollbackFile)) | 是故障恢复的重要机制 |
7.5 数据恢复（Restore）
| 参数/方法 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.restore.enable | true | 启用恢复功能 | 默认开启 | 无需额外配置 |
| hoodie.restore.commits | 提交时间列表 | 指定要恢复的 commit 时间点 | .option(“hoodie.restore.commits”, “20250101000000,20250101010000”) | 按顺序恢复 |
| hoodie.restore.parallelism | 整数 | 恢复操作的并行度 | .option(“hoodie.restore.parallelism”, “5”) | 根据数据量调整 |
| 触发方式 | 手动执行 | 通过 hudi-cli 或 Spark 作业 | RestoreTool 命令行工具 | 不支持自动触发 |
| 恢复原理 | 重放 timeline | 将表状态恢复到指定 commit 时间点 | 类似 Git 的 reset 操作 | 不会删除后续数据，而是创建新分支 |
| 适用场景 | 误写入、逻辑错误修复 | 恢复到错误发生前的状态 | 需提前保留 savepoint 或未被归档的元数据 |
第8章：Hudi 与生态集成
8.1 Hudi 与 Hive 集成
| 参数/方法 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| hoodie.datasource.hive_sync.enable | true | 启用 Hive 元数据同步 | .option(“hoodie.datasource.hive_sync.enable”, “true”) | 必须开启才能在 Hive 中查询 |
| hoodie.datasource.hive_sync.table | 表名 | 指定同步到 Hive 的表名 | .option(“hoodie.datasource.hive_sync.table”, “hudi_users”) | 默认为 Hudi 表名 |
| hoodie.datasource.hive_sync.partition_fields | 分区字段 | 指定 Hive 分区字段 | .option(“hoodie.datasource.hive_sync.partition_fields”, “dt”) | 必须与写入配置一致 |
| hoodie.datasource.hive_sync.jdbcurl | JDBC URL | Hive Metastore 地址 | .option(“hoodie.datasource.hive_sync.jdbcurl”, “jdbc:hive2://hivemetastore:10000”) | 需网络可达 |
| hoodie.datasource.hive_sync.username | 用户名 | 访问 Hive 的用户名 | .option(“hoodie.datasource.hive_sync.username”, “hive”) | 权限需足够 |
| 同步方式 | 写入时自动同步或异步工具 | 可通过 HiveSyncTool 手动同步 | spark-submit —class org.apache.hudi.utilities.HiveSyncTool … | 推荐自动同步 |
8.2 Hudi 与 Spark 集成
| 集成方式 | 说明 | 代码示例 | 注意事项 |
|---|---|---|---|
| Spark Datasource API | 使用 format(“hudi”) 写入 | df.write.format(“hudi”).save(path) | 最常用方式 |
| Spark SQL | 通过注册临时视图执行 SQL | spark.sql(“INSERT INTO hudi_table SELECT …”) | 需先创建表 |
| Spark Structured Streaming | 流式写入 Hudi 表 | streamDF.writeStream.format(“hudi”).start() | 支持微批处理 |
| 依赖包 | hudi-spark3-bundle | Maven 坐标：org.apache.hudi:hudi-spark3-bundle_2.12:x.x.x | 根据 Spark 版本选择 |
| 读取方式 | DataFrame 或 SQL | spark.read.format(“hudi”).load(path) | 支持 snapshot、incremental 查询 |
| 性能优化 | 启用 Kryo、AQE | spark.conf.set(“spark.sql.adaptive.enabled”, true) | 显著提升性能 |
8.3 Hudi 与 Flink 集成（Flink CDC）
| 参数/方法 | 语法 | 用途 | 代码示例 | 注意事项 |
|---|---|---|---|---|
| FlinkHoodieSink | Flink Sink 实现 | 将 Flink 流写入 Hudi 表 | FlinkHoodieSink.forRowData(rowType).build() | 需引入 hudi-flink-bundle |
| write.operation | upsert / insert | 指定写入操作类型 | .withWriteOperation(WriteOperation.UPSERT) | 默认为 upsert |
| index.bootstrap.enabled | true/false | 是否启用索引导入（首次加载） | .withIndexBootstrapEnabled(true) | 加速首次写入 |
| compaction.async.enabled | true | 异步压缩 | .withAsyncCompactionEnabled(true) | 避免阻塞写入流 |
| CDC 源集成 | Debezium + Flink CDC | 读取数据库变更日志 | MySQLSource.builder().hostname(“localhost”).databaseList(“db”).build() | 实现端到端 CDC |
| 状态后端 | RocksDB | 推荐使用 RocksDB 状态后端 | env.setStateBackend(new RocksDBStateBackend(…)) | 支持大状态 |
8.4 Hudi 与 Presto/Trino 查询引擎集成
| 配置项 | 说明 | 配置方式 | 注意事项 |
|---|---|---|---|
| catalog.hudi.properties | Catalog 配置文件 | 创建 etc/catalog/hudi.properties | Presto/Trino 插件方式集成 |
| connector.name | 固定为 hudi | connector.name=hudi | 必须设置 |
| hive.metastore.uri | Hive Metastore 地址 | hive.metastore.uri=thrift://hms:9083 | Hudi 表需已同步到 Hive |
| 查询模式 | Snapshot Only | Presto/Trino 仅支持 snapshot 查询 | SELECT * FROM hudi_db.table |
| 版本兼容性 | Hudi 0.12+ | 推荐使用 Hudi 0.12 或更高版本 | 早期版本可能存在兼容问题 |
| 性能 | 依赖 Parquet 扫描 | 可利用分区剪枝、谓词下推 |
8.5 Hudi 与 Delta Lake/Iceberg 对比
| 对比维度 | Hudi | Delta Lake | Iceberg | 注意事项 |
|---|---|---|---|---|
| 开源归属 | Apache 项目 | Databricks 主导（Linux 基金会） | Apache 项目 | Hudi 和 Iceberg 更中立 |
| 写入模型 | 支持 COW 和 MOR | 主要 COW（支持 Z-Order 等优化） | COW 为主 | Hudi 的 MOR 更适合高吞吐更新 |
| 增量查询 | 原生支持，基于 timeline | 支持 CHANGES IN TIME RANGE | 支持 changelog 模式 | 三者均支持 |
| 索引机制 | 内置多种索引（Bloom、HBase） | 依赖数据跳过（Data Skipping） | 依赖 Manifest 文件 | Hudi 索引更灵活 |
| 流式写入 | Flink、Spark Streaming 支持好 | Spark Streaming 集成最佳 | Flink 支持较好 | 各有优势 |
| 生态兼容 | Hive、Presto、Flink、Spark | Spark 生态无缝集成 | Presto/Trino 支持最好 | 根据技术栈选择 |
| 社区活跃度 | 高 | 高（商业支持强） | 高 | 均为活跃项目 |
| 存储格式 | Parquet + Avro（log） | Parquet 为主 | Parquet/ORC/Avro | 均基于开放格式 |
第9章：Hudi 实战案例
9.1 实时数仓中的 Upsert 场景
| 项目 | 内容 |
|---|---|
| 场景描述 | 用户行为日志实时写入，订单状态持续更新，需保证 record key 唯一性 |
| 表类型选择 | MERGE_ON_READ（高写入吞吐）或 COPY_ON_WRITE（强一致性读） |
| 写入操作 | upsert |
| 关键参数配置 | .option(“hoodie.table.type”, “MERGE_ON_READ”) / .option(“hoodie.datasource.write.operation”, “upsert”) / .option(“hoodie.datasource.write.recordkey.field”, “order_id”) / .option(“hoodie.datasource.write.partitionpath.field”, “dt”) / .option(“hoodie.upsert.shuffle.parallelism”, “20”) / .option(“hoodie.combine.before.upsert”, “true”) |
| 代码示例 | df.write.format(“hudi”).option(“hoodie.table.name”, “orders_realtime”).option(“hoodie.datasource.write.table.type”, “MERGE_ON_READ”).option(“hoodie.datasource.write.operation”, “upsert”).option(“hoodie.datasource.write.recordkey.field”, “order_id”).option(“hoodie.datasource.write.partitionpath.field”, “dt”).option(“hoodie.upsert.shuffle.parallelism”, “20”).mode(“append”).save(“/data/hudi/orders”) |
| 注意事项 | 使用 GLOBAL_BLOOM 索引避免跨分区重复；开启 combine.before.upsert 减少索引查询；MOR 表需配置异步压缩保障读性能 |
9.2 增量同步数据到数据湖
| 项目 | 内容 |
|---|---|
| 场景描述 | 从 MySQL CDC 捕获变更，增量写入 Hudi 数据湖 |
| 技术栈 | Flink CDC → Kafka → Flink → Hudi |
| Hudi 写入模式 | upsert 或 insert（根据是否需要更新） |
| 关键参数配置 | .option(“hoodie.datasource.write.operation”, “upsert”) / .option(“hoodie.datasource.write.table.type”, “COPY_ON_WRITE”) / .option(“hoodie.cleaner.policy”, “KEEP_LATEST_COMMITS”) / .option(“hoodie.cleaner.commits.retained”, “10”) / .option(“hoodie.datasource.write.hive_style_partitioning”, “true”) |
| 代码示例（Flink） | DataStream stream = env.addSource(mySqlSource); HoodieFlinkStreamer streamer = HoodieFlinkStreamer.forRowData(stream).withWriteConfig(HoodieWriteConfig.newBuilder().withPath(“/data/hudi/cdc_table”).withTableName(“cdc_table”).withWriteOperation(WriteOperation.UPSERT).build()).build(); streamer.start(); |
| 注意事项 | 确保 Flink Checkpoint 与 Hudi Commit 对齐；使用 WriteClient 控制 commit 频率；监控延迟与 backpressure |
9.3 软删除与硬删除实现
| 项目 | 内容 |
|---|---|
| 软删除实现 | 标记删除，逻辑删除 |
| 配置参数 | .option(“hoodie.datasource.write.operation”, “delete”) / .option(“hoodie.datasource.write.payload.class”, “org.apache.hudi.EmptyHoodieRecordPayload”) |
| 代码示例（软删除） | deletedDf.write.format(“hudi”).option(“hoodie.datasource.write.operation”, “delete”).option(“hoodie.datasource.write.payload.class”, “org.apache.hudi.EmptyHoodieRecordPayload”).option(“hoodie.delete.shuffle.parallelism”, “10”).mode(“append”).save(“/data/hudi/table”) |
| 硬删除实现 | 物理删除，需配合清理策略 |
| 配置参数 | .option(“hoodie.cleaner.policy”, “KEEP_LATEST_FILE_VERSIONS”) / .option(“hoodie.cleaner.fileversions.retained”, “1”) |
| 注意事项 | 软删除后数据仍存在于文件中，查询时不可见；硬删除需等待 clean 操作完成；删除操作需确保 record key 正确 |
9.4 分区表与非分区表写入对比
| 对比维度 | 分区表 | 非分区表 | 注意事项 |
|---|---|---|---|
| 写入性能 | 中等（需路由到分区） | 高（无分区逻辑） | 分区越多，写入开销越大 |
| 读取性能 | 高（支持分区剪枝） | 低（全表扫描） | 分区字段应具高区分度 |
| 小文件问题 | 更严重（每个分区可能产生小文件） | 相对较少 | 需配置 small.file.size 和并行度 |
| 配置示例 | .option(“hoodie.datasource.write.partitionpath.field”, “dt”) | 不设置 partitionpath.field | 默认为非分区表 |
| 适用场景 | 按时间、地域等维度查询频繁 | 数据量小，无明确分区逻辑 | 建议生产环境使用分区表 |
9.5 性能调优实战
| 调优方向 | 配置建议 | 效果 | 注意事项 |
|---|---|---|---|
| 并行度优化 | hoodie.upsert.shuffle.parallelism=20 / hoodie.bulkinsert.shuffle.parallelism=50 | 提升写入吞吐 | 并行度 ≈ 目标文件数 |
| 文件大小控制 | hoodie.parquet.max.file.size=536870912 (512MB) / hoodie.parquet.small.file.size=134217728 (128MB) | 减少小文件，提升读性能 | 避免过大影响调度 |
| 索引优化 | hoodie.index.type=GLOBAL_BLOOM / hoodie.bloom.filter.type=DYNAMIC | 加速 upsert 去重 | 全局索引消耗更多内存 |
| 压缩与清理 | hoodie.compact.inline=true / hoodie.clean.async=true | 保障 MOR 表读性能 | 控制 max.delta.commits |
| Spark 优化 | spark.sql.adaptive.enabled=true / spark.serializer=org.apache.spark.serializer.KryoSerializer | 提升执行效率 | 建议在生产环境启用 |
第10章：Hudi 高级特性与最佳实践
10.1 多版本并发控制（MVCC）
| 项目 | 内容 |
|---|---|
| 机制说明 | Hudi 通过时间轴（Timeline）实现 MVCC，保证读写隔离 |
| 核心组件 | .hoodie/ 目录下的 commit、deltacommit、clean、rollback 等 instant 文件 |
| 读写隔离 | 读操作基于某个 commit instant 的快照，写操作追加新 instant，互不阻塞 |
| 一致性保证 | Snapshot Query 提供快照隔离级别 |
| 注意事项 | 时间轴归档需合理配置，避免元数据膨胀；并发写入需协调，避免冲突（通常由调度系统控制） |
10.2 Schema Evolution 支持
| 项目 | 内容 |
|---|---|
| 支持操作 | 字段新增、字段默认值、字段重命名（需配置） |
| 启用方式 | 默认开启，无需额外配置 |
| 配置参数 | hoodie.datasource.write.schema.evolution (true/false) / hoodie.avro.schema.validate (验证 schema 兼容性) |
| 代码示例 | 原 schema: {“name”: “id”, “type”: “int”} → 新数据: {“id”: 1, “name”: “Alice”} → 自动扩展 schema |
| 注意事项 | 不支持字段类型变更（如 int → string）；建议使用兼容的 schema 变更策略；查询时旧文件会填充 null |
10.3 异常处理与监控
| 项目 | 内容 |
|---|---|
| 常见异常 | 写入失败、小文件过多、压缩延迟、元数据损坏 |
| 监控指标 | Commit Duration / Ingestion Rate / Number of Files / Log-to-Base Ratio (MOR) / Timeline Age |
| 监控工具 | Prometheus + Grafana、Hudi CLI、Spark UI |
| 告警建议 | Commit 超时告警；Clean/Compaction 延迟告警；存储空间突增告警 |
| 日志分析 | 查看 .hoodie/<instant>/*.commit 文件分析写入详情 |
10.4 生产环境部署建议
| 项目 | 建议内容 |
|---|---|
| 集群规划 | 写入与查询分离（如 Flink 写，Presto 读）；资源隔离，避免相互影响 |
| 表设计 | 合理选择 COW/MOR；使用分区 + 全局索引；预设文件大小与并行度 |
| 运维策略 | 定期归档与清理；启用 Metadata Table 加速 listing；备份 .hoodie 元数据 |
| 版本选择 | 使用稳定版本（如 0.13.x、0.14.x），避免使用 RC 版本 |
| 高可用 | Hive Metastore HA；存储系统（HDFS/S3）冗余 |
10.5 安全性与权限控制
| 项目 | 说明 |
|---|---|
| 认证机制 | 支持 Kerberos（HDFS）、IAM（S3）、SAS（ADLS） |
| 授权机制 | 依赖底层文件系统 ACL 或 Ranger/Sentry |
| 数据加密 | 传输加密（SSL/TLS）；静态加密（KMS + HDFS/S3） |
| 字段级安全 | 通过视图或查询层过滤敏感字段 |
| 审计日志 | 记录写入、删除、回滚等操作，集成到统一审计平台 |