一、前言：数据湖的困境

1.1 传统数据湖的痛点

1.2 Hudi 的诞生

Hudi = Hadoop Upserts Deletes and Incrementals

Uber 的工程师发现：打车订单需要频繁更新状态（下单→接单→进行中→完成），传统数据湖完全搞不定！

二、Hudi 核心概念

2.1 整体架构

2.2 Timeline：时间线机制

Timeline 记录了表的所有操作历史：

2.3 文件布局

hudi_table/
├── .hoodie/                      # 元数据目录
│   ├── hoodie.properties        # 表属性
│   ├── 20240115100000.commit    # 提交记录
│   ├── 20240115110000.commit
│   └── ...
├── 2024/01/15/                   # 分区目录
│   ├── file_1.parquet           # 基础文件
│   ├── file_2.parquet
│   ├── .file_1.log.1            # 日志文件 (MOR)
│   └── .file_1.log.2
└── 2024/01/16/
    └── ...

2.4 File Group 与 File Slice

核心概念：

三、两种表类型：COW vs MOR

3.1 Copy-On-Write (COW)

COW 特点：

3.2 Merge-On-Read (MOR)

MOR 特点：

3.3 COW vs MOR 对比

3.4 选型建议

四、索引机制

4.1 为什么需要索引？

4.2 索引类型

4.3 Bloom Index 原理

Bloom Index 配置：

# Bloom Filter 配置
hoodie.bloom.index.filter.type=DYNAMIC_V0
hoodie.bloom.index.filter.dynamic.max.entries=100000
hoodie.index.bloom.num_entries=60000
hoodie.index.bloom.fpp=0.000000001

4.4 索引选择建议

五、增量查询：Hudi 的杀手锏

5.1 三种查询模式

5.2 增量查询示例

Spark 增量查询代码：

// 增量读取 Hudi 表
val incrementalDF = spark.read
  .format("hudi")
  .option("hoodie.datasource.query.type", "incremental")
  .option("hoodie.datasource.read.begin.instanttime", "20240115100000")
  .option("hoodie.datasource.read.end.instanttime", "20240115120000")
  .load("/path/to/hudi_table")

// 处理增量数据
incrementalDF.show()

// 获取本次消费的最新位点
val latestCommit = incrementalDF
  .select("_hoodie_commit_time")
  .agg(max("_hoodie_commit_time"))
  .first().getString(0)

Flink 增量消费：

-- Flink SQL 流式读取 Hudi
CREATE TABLE hudi_orders (
    order_id BIGINT,
    user_id BIGINT,
    amount DECIMAL(10,2),
    order_time TIMESTAMP(3),
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector' = 'hudi',
    'path' = 's3://bucket/orders',
    'table.type' = 'MERGE_ON_READ',
    'read.streaming.enabled' = 'true',
    'read.streaming.check-interval' = '10'  -- 10秒检查一次新数据
);

-- 实时消费变化数据
SELECT * FROM hudi_orders;

5.3 增量 ETL 架构

六、Compaction 与 Clustering

6.1 Compaction（压缩合并）

Compaction 策略：

# 同步 Compaction（写入时执行）
hoodie.compact.inline=true
hoodie.compact.inline.max.delta.commits=5

# 异步 Compaction（独立任务执行）
hoodie.compact.inline=false
# 需要单独运行 Compaction 任务

6.2 Clustering（聚簇优化）

Clustering 的好处：

6.3 Clean（清理旧版本）

# 清理配置
hoodie.cleaner.policy=KEEP_LATEST_COMMITS
hoodie.cleaner.commits.retained=3

# 或者按时间保留
hoodie.cleaner.policy=KEEP_LATEST_BY_HOURS
hoodie.cleaner.hours.retained=24

七、实战：Hudi 表操作

7.1 创建 Hudi 表

// Spark 创建 Hudi 表
import org.apache.hudi.QuickstartUtils._
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.config.HoodieWriteConfig._

val tableName = "hudi_orders"
val basePath = "s3://bucket/hudi/orders"

// 初始数据
val initialDF = spark.createDataFrame(Seq(
  (1L, 1001L, 100.00, "2024-01-15 10:00:00"),
  (2L, 1002L, 200.00, "2024-01-15 10:01:00"),
  (3L, 1003L, 150.00, "2024-01-15 10:02:00")
)).toDF("order_id", "user_id", "amount", "order_time")

// 写入 Hudi 表
initialDF.write
  .format("hudi")
  .options(Map(
    TABLE_NAME.key -> tableName,
    RECORDKEY_FIELD.key -> "order_id",           // 主键
    PRECOMBINE_FIELD.key -> "order_time",        // 预合并字段
    PARTITIONPATH_FIELD.key -> "",               // 分区字段（空=不分区）
    TABLE_TYPE.key -> "MERGE_ON_READ",           // 表类型
    "hoodie.datasource.write.operation" -> "upsert"
  ))
  .mode("overwrite")
  .save(basePath)

7.2 Upsert 操作

// 更新数据（Upsert）
val updateDF = spark.createDataFrame(Seq(
  (1L, 1001L, 120.00, "2024-01-15 11:00:00"),  // 更新
  (4L, 1004L, 300.00, "2024-01-15 11:01:00")   // 新增
)).toDF("order_id", "user_id", "amount", "order_time")

updateDF.write
  .format("hudi")
  .options(Map(
    TABLE_NAME.key -> tableName,
    RECORDKEY_FIELD.key -> "order_id",
    PRECOMBINE_FIELD.key -> "order_time",
    "hoodie.datasource.write.operation" -> "upsert"
  ))
  .mode("append")
  .save(basePath)

7.3 Delete 操作

// 删除数据
val deleteDF = spark.createDataFrame(Seq(
  (2L, 1002L, 0.00, "2024-01-15 12:00:00")  // 要删除的记录
)).toDF("order_id", "user_id", "amount", "order_time")

deleteDF.write
  .format("hudi")
  .options(Map(
    TABLE_NAME.key -> tableName,
    RECORDKEY_FIELD.key -> "order_id",
    PRECOMBINE_FIELD.key -> "order_time",
    "hoodie.datasource.write.operation" -> "delete"
  ))
  .mode("append")
  .save(basePath)

7.4 时间旅行查询

// 查询某个时间点的快照
val historicalDF = spark.read
  .format("hudi")
  .option("as.of.instant", "20240115100000")  // 指定时间点
  .load(basePath)

historicalDF.show()

// 查询两个时间点之间的变化
val changesDF = spark.read
  .format("hudi")
  .option("hoodie.datasource.query.type", "incremental")
  .option("hoodie.datasource.read.begin.instanttime", "20240115100000")
  .option("hoodie.datasource.read.end.instanttime", "20240115120000")
  .load(basePath)

八、Hudi + Flink 实时入湖

8.1 架构图

8.2 Flink SQL 示例

-- 1. 创建 Kafka 源表（CDC 数据）
CREATE TABLE kafka_orders (
    order_id BIGINT,
    user_id BIGINT,
    amount DECIMAL(10,2),
    order_status STRING,
    order_time TIMESTAMP(3),
    op_type STRING,  -- CDC 操作类型
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector' = 'kafka',
    'topic' = 'orders_cdc',
    'properties.bootstrap.servers' = 'kafka:9092',
    'format' = 'debezium-json'
);

-- 2. 创建 Hudi Sink 表
CREATE TABLE hudi_orders (
    order_id BIGINT,
    user_id BIGINT,
    amount DECIMAL(10,2),
    order_status STRING,
    order_time TIMESTAMP(3),
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector' = 'hudi',
    'path' = 's3://bucket/hudi/orders',
    'table.type' = 'MERGE_ON_READ',
    'hoodie.datasource.write.recordkey.field' = 'order_id',
    'hoodie.datasource.write.precombine.field' = 'order_time',
    'write.operation' = 'upsert',
    'write.tasks' = '4',
    'compaction.tasks' = '2',
    'compaction.async.enabled' = 'true',
    'compaction.delta_commits' = '5'
);

-- 3. 实时写入
INSERT INTO hudi_orders
SELECT order_id, user_id, amount, order_status, order_time
FROM kafka_orders;

九、最佳实践

9.1 配置优化

9.2 关键配置

# ===== 写入配置 =====
# 目标文件大小
hoodie.parquet.max.file.size=134217728  # 128MB
hoodie.parquet.small.file.limit=104857600  # 100MB

# 写入并行度
hoodie.insert.shuffle.parallelism=200
hoodie.upsert.shuffle.parallelism=200

# ===== 索引配置 =====
hoodie.index.type=BLOOM
hoodie.bloom.index.parallelism=200

# ===== Compaction 配置 =====
hoodie.compact.inline=false
hoodie.compact.inline.max.delta.commits=5
hoodie.compaction.strategy=org.apache.hudi.table.action.compact.strategy.LogFileSizeBasedCompactionStrategy

# ===== 清理配置 =====
hoodie.cleaner.policy=KEEP_LATEST_COMMITS
hoodie.cleaner.commits.retained=10

9.3 常见问题

十、总结

10.1 Hudi 核心价值

10.2 适用场景

10.3 一句话总结

Hudi = 让数据湖支持增删改 + 增量处理 + 时间旅行

如果你需要在数据湖上实现实时更新和增量消费，Hudi 是目前最成熟的选择之一。