Apache Hudi 学习笔记 06：Hudi 写入流程与写操作类型详解

hudi

发布日期: 2025-07-29

文章字数: 6k

阅读时长: 24 分

阅读次数:

前面讲完 COW、MOR 和查询类型以后，再看写入操作就清楚多了。Hudi 写入不是简单把 DataFrame 保存成 Parquet，而是要根据 record key、partition path、precombine 字段、索引定位和表类型，决定这批数据最终是插入、更新、删除，还是覆盖某个分区。

Hudi 的写操作很多，但学习阶段最核心的是这几类：upsert、insert、bulk_insert、insert_overwrite、insert_overwrite_table 和 delete。这些操作的差异，决定了是否查索引、是否允许更新、是否重写文件组、是否产生 replace commit。

1. 写入语义和操作类型

1.1 背景问题

假设我们要把订单数据写入 Hudi：

order_id=3001, dt=2025-07-28, status=CREATED
order_id=3002, dt=2025-07-28, status=CREATED
order_id=3003, dt=2025-07-29, status=CREATED

这里的 dt 是模拟业务分区日期，用于演示跨分区写入和分区覆盖，不代表文章实际写作日期。

后续会发生几种变化：

初始化导入一批历史数据。
追加一个新订单 3004。
把 3001 从 CREATED 更新成 PAID。
删除错误订单 3002。
重算 dt=2025-07-29 分区，只保留修正后的 3999。

这些场景都叫“写入”，但语义完全不同。如果全部都用 upsert，能跑，但不一定最高效；如果该用 upsert 的地方用了 insert，又可能引入重复数据或错误结果。

所以这一篇先把写操作语义讲清楚。

1.2 写入前的几个关键概念

Hudi 写入至少要理解四个字段或参数。

参数	作用
`hoodie.datasource.write.recordkey.field`	Hudi record key / 业务主键字段，用来决定同一条业务记录如何识别；它不是传统数据库里的强制唯一约束
`hoodie.datasource.write.partitionpath.field`	分区路径，决定数据落在哪个分区目录
`hoodie.datasource.write.precombine.field`	同一批数据中同 key 冲突时保留哪条
`hoodie.datasource.write.operation`	写操作类型，比如 `upsert`、`insert`、`delete`

这几个参数共同决定 HoodieKey 和写入路径。

对分区表可以先理解为：

HoodieKey = record key + partition path

对非分区表，partition path 通常为空或由非分区 KeyGenerator 处理。

对于分区表，record key 和 partition path 都很重要。非全局索引通常把 partition path 作为定位范围的一部分；如果同一个 record key 的分区字段会变化，就可能在新旧分区之间产生重复或旧记录残留，需要考虑 Global Index、Record Index、update partition path 相关配置，或由业务侧保证分区字段稳定。

写操作选择可以先按业务语义判断，而不是先背参数：

Hudi 写操作选型

这张图主要说明写操作应先按业务语义选择：初始化导入用 bulk_insert，纯追加用 insert，有变更用 upsert，删除用 delete，分区修复用 insert_overwrite。

这张图里最重要的是几条分支：

历史全量初始化，优先考虑 bulk_insert，它更关注吞吐和文件组织，不适合拿来处理日常更新。
纯追加事件表，如果上游已经保证不重复，可以使用 insert，减少索引定位成本。
CDC 或状态表，只要同一个业务 key 可能再次出现，就应该使用 upsert。
删除事件用 delete 表达，不要用手工过滤旧数据的方式绕过 Timeline。
分区重算、数据修复更适合 insert_overwrite，它表达的是替换某个范围的有效文件视图。

1.3 UPSERT

upsert 是 Hudi 最典型的写操作，也是 CDC 入湖最常用的写操作。

它的语义是：

key 已存在 -> update
key 不存在 -> insert

图示如下：

UPSERT 写入流程

这张图主要说明 upsert 会先通过 record key、partition path 和索引判断记录是否已存在，再分别进入 update path 或 insert path。

upsert 的大致流程：

根据 record key 和 partition path 生成 HoodieKey。
按 precombine.field 处理同一批数据中相同 key 的多条记录。
使用索引查找 key 是否已经存在。
给记录打 location。
已存在的记录进入 update path。
不存在的记录进入 insert path。
提交 Timeline Instant。

在 COW 表里，update path 通常会重写 Base File；在 MOR 表里，update path 可以追加 Log File。

把 COW 和 MOR 拆开看，upsert 的文件动作大致是这样：

表类型	update path	insert path
COW	找到 key 所在最新 FileSlice，读取旧 Base File，合并后写新的 Base File	优先写入当前分区可用的 Small File；没有合适小文件时创建新的 FileGroup
MOR	找到 key 所在 FileGroup 后，典型情况下会把更新写入 Log File，后续由 Snapshot Query 或 compaction 合并	新增记录可能写入 Base File，也可能在特定写入路径下进入 Log File；实际形态受写入引擎、small file 策略、索引和 compaction 状态影响

这里和上一篇讲 MOR 时的结论一致：MOR 不等于所有数据都一定先进 Log File。写入引擎、索引类型、small file 策略都会影响真实文件形态。

upsert 适合：

MySQL CDC 入湖。
维表更新。
状态表同步。
需要同时处理新增和变更的场景。

代价是它需要做索引定位，所以比纯追加写入更重。

1.4 INSERT

insert 更偏纯追加。

它的语义是：把输入记录作为新记录写入表，不按 upsert 那样完整执行更新定位。

常见配置：

hoodie.datasource.write.operation=insert

insert 适合：

日志明细表。
事件流。
明确不会更新的事实明细。
已经在上游保证无重复的新数据。

它的优势是更快，少了更新定位的成本。缺点是如果输入里包含已有 key，就可能带来重复或语义错误。因为 insert 不会像 upsert 那样完整执行已有 key 的更新定位，所以已有 key 可能被当成新记录写入，最终在快照中造成重复或非预期结果，具体表现还会受索引和去重参数影响。对于设置了 record key 的 Hudi 表，Spark SQL 的 insert strict / non-strict 还有更细的语义差异，后面 Spark SQL 篇会单独展开。

insert 不是完全什么都不做。它通常跳过索引定位，但仍可能根据配置做批内合并或 small file 处理：

表类型	INSERT 写入特点
COW	不执行 `upsert` 那种已有记录更新定位，主要按追加写入和 small file 策略组织文件
MOR	不执行 `upsert` 那种已有记录更新定位；实际写入 Base File 还是 Log File，要看写入路径、small file 策略和表服务状态

我的习惯是：如果业务语义是“这个 key 可能已经存在”，就不要为了速度随便用 insert。

1.5 BULK_INSERT

bulk_insert 主要用于初始化大批量导入。

常见配置：

hoodie.datasource.write.operation=bulk_insert

官方文档对它的定位很明确：先用 bulk_insert 加载新表的大量初始数据，后续再用 upsert 或 insert 持续写入。它的写入路径更适合大输入，通常通过更偏排序 / 批量写入的方式提升初始导入吞吐，避免普通 insert / upsert 在大规模初始导入时的高内存压力。

bulk_insert 适合：

首次把历史 Hive 表导入 Hudi。
初始化 ODS 大表。
一次性大批量导入，不需要逐条更新旧数据。

不适合：

CDC 增量更新。
有大量重复 key 需要合并的场景。
需要严格更新语义的写入。

bulk_insert 还要注意两个工程细节。

第一个是排序。初始化大表时，如果输入数据完全乱序，写入端会频繁切换分区和文件句柄，容易制造很多小文件。批量导入时通常会尽量按 partition path 排序，减少 file handle 来回切换。

第二个是并发。写入并发越高，吞吐可能越好，但初始文件数也可能变多。可以粗略理解成：写入并发越高，初始输出文件和 FileGroup 数量越容易增加；每个文件写到目标大小后再 roll over 出新的文件。后面调优篇会再讲 write.tasks、文件大小和 small file 的关系。

简单说：

bulk_insert = 初始化导入优先
upsert = 持续变更优先
insert = 纯追加优先

1.6 DELETE

Hudi 的删除不是直接在 Parquet 文件中删除一行，而是通过新提交让这条记录在新的表快照中不可见。

图示如下：

DELETE 写入流程

这张图主要说明 Hudi delete 不是直接修改旧 Parquet 文件，而是通过新的提交改变最新快照中的记录可见性。

Spark DataSource 可以直接使用 delete 操作：

hoodie.datasource.write.operation=delete

输入数据至少要能构造出被删除记录的 record key 和 partition path。为了让合并逻辑稳定，CDC delete event 也应带有能正确排序的 precombine 字段，确保删除事件不会因为 ts 小于旧记录而被合并逻辑忽略。比如：

order_id=3002, dt=2025-07-28

删除流程大致是：

根据 key 定位已有记录。
在 COW 表中重写新的 Base File，让该 key 不再出现。
在 MOR 表中可以写入删除标记，读取或 compaction 时合并。
提交新的 Instant。

还有一种常见写法是 _hoodie_is_deleted 字段。也就是在普通 upsert 数据里带一个删除标记：

order_id=3002, dt=2025-07-28, _hoodie_is_deleted=true

这种方式在 CDC 流里很常见，因为 Debezium / Canal 这类源端通常会带操作类型，下游可以把 delete event 转成 Hudi 的删除语义。使用 _hoodie_is_deleted 时，写入操作通常仍是 upsert，但输入记录中带删除标记，Hudi 在合并时把它解释为删除语义。具体字段是否生效与 Hudi 版本、payload 和 DataSource 写入路径有关，实测文章中要以当前 Hudi 1.0.2 行为为准。

删除还要区分两层含义。

删除方式	含义	适合场景
业务软删	保留记录，只把 `is_deleted`、`status` 等业务字段改成删除状态	下游需要看到删除状态或保留审计信息
Hudi delete	通过 `delete` 操作或 `_hoodie_is_deleted=true` 让记录在最新快照中不可见	撤单、脏数据清理、CDC delete event 入湖
旧版 Payload 删除	使用类似 `EmptyHoodieRecordPayload` 的 payload 表达删除	旧 DataSource 代码或自定义 payload 场景

本文主线使用 delete 和 _hoodie_is_deleted。Payload 删除更多用于旧代码兼容或自定义写入逻辑，学习阶段不作为优先路线。不要把“把业务字段置空”误认为 Hudi 的 delete。前者仍然是一条可见记录，后者改变的是 Hudi 表快照中的可见性。

1.7 INSERT_OVERWRITE

insert_overwrite 用于覆盖目标分区。

它不是手工删除 HDFS 目录再重写，而是通过 Hudi 的 replace commit 语义，让旧 FileGroup 失效，用新的 FileGroup 替换目标分区。

图示如下：

INSERT_OVERWRITE 分区覆盖流程

这张图主要说明 insert_overwrite 会通过 replacecommit 替换目标分区的有效 FileGroup，而不是手工删除 HDFS 目录。

常见配置：

hoodie.datasource.write.operation=insert_overwrite

在 Hudi 1.0.2 Spark DataSource 中，insert_overwrite 会重写输入 DataFrame 中出现的分区。hoodie.datasource.write.partitions.to.delete 是 delete_partition 场景的配置，不是本文 insert_overwrite 示例的主线配置。不同版本和 Spark SQL 写法中，分区覆盖范围也可能由输入数据分区或 SQL PARTITION 子句决定。

适合场景：

某天分区口径修复。
维表某个分区重算。
批处理回补数据。
希望保留 Hudi Timeline 记录，而不是手动删目录。

还有一个相关操作是 insert_overwrite_table，用于覆盖整张表。insert_overwrite_table 表达的是 Hudi 表级逻辑覆盖语义，不要把它简单等同于手工删除表目录再重写。Spark 的 mode("overwrite") 与 Hudi operation 组合使用时，具体行为要以当前 Hudi 版本和写入配置为准。

insert_overwrite 的关键点是：它会创建新的 FileGroup 集合，并把目标范围内旧 FileGroup 标记为失效。这样做有几个好处：

不需要手工删除 HDFS 目录，Timeline 能记录替换过程。
对 COW 和 MOR 的处理路径比较统一，不需要强行干扰 MOR compaction。
后续写入和文件视图构造可以根据 Timeline 识别旧 FileGroup 已被替换，从而避免把当前有效视图继续建立在旧文件组之上。具体索引状态如何更新，要结合索引类型和当前版本实现理解。
Cleaner 可以在安全窗口之后再清理旧文件，避免正在运行的旧查询被直接打断。

但它也有一个容易误解的地方：旧 Parquet 文件在磁盘上可能还存在，只是当前文件视图已经不再选择它们。排查问题时不能只看 HDFS 目录里有没有旧文件，还要结合 replacecommit 和 Snapshot Query 的结果判断。

2. 写入策略和实测验证

2.1 Key 生成策略

Hudi 写入绕不开 Key Generator。

最常见的是简单 key：

hoodie.datasource.write.recordkey.field=order_id
hoodie.datasource.write.partitionpath.field=dt

如果业务主键是多字段，可以配置组合 key：

hoodie.datasource.write.recordkey.field=tenant_id,order_id
hoodie.datasource.write.partitionpath.field=dt

多字段 record key 或多字段 partition path 时，建议显式配置合适的 KeyGenerator，例如 ComplexKeyGenerator 或对应版本推荐的 keygenerator.class。具体类名和参数以 Hudi 1.0.2 文档为准。

分区字段也可以是多字段组合。例如按日期和地区分区：

hoodie.datasource.write.partitionpath.field=dt,region
hoodie.datasource.write.hive_style_partitioning=true

开启 Hive style partitioning 后，目录会更接近 Hive 常见形态：

dt=2025-07-28/region=cn/

如果是不分区表，通常要使用 NonpartitionedKeyGenerator 或 Hudi 当前版本推荐的非分区配置，并确保 partitionpath.field 为空或不参与路径生成。不要只把 partitionpath.field 留空而不确认 KeyGenerator 行为。实际项目里是否分区，要看查询过滤条件、写入规模、分区字段是否稳定。

本文实测表为了简化只使用 dt 单字段分区，并显式开启 Hive style partitioning，方便在 HDFS 上看到 dt=2025-07-29 这类目录。多字段分区和 Hive Sync 后续 Spark SQL 篇再结合实际目录展示。

我一般会先确认三件事：

record key 是否真的唯一。
partition path 是否会变化。
下游查询是否经常按 partition path 过滤。

Key 设计错了，后面索引、更新、删除、Hive Sync 都会跟着难受。

2.2 Small File 策略

Hudi 写入时还会处理 small file。

如果每次增量都创建新文件，很快会形成大量小文件。Hudi 在 insert path 中会尽量把新记录写入已有小文件对应的 FileGroup，减少小文件数量。

相关参数后面调优篇会详细讲，这里先记住两个：

hoodie.parquet.max.file.size=134217728
hoodie.parquet.small.file.limit=104857600

这两个只是理解 small file 的入门参数，真实调优还要结合写入并发、记录大小估计、shuffle 并发、文件大小、表类型和写入操作一起看，后续调优篇再展开。

理解 small file 策略后，就能解释为什么 insert 不一定总是创建新文件，Hudi 可能会把新记录合并进小文件范围内的 FileGroup。

2.3 Spark 示例：连续执行多种写操作

本次创建一张 COW 表：

hdfs://localhost:9000/hudi/demo_write_ops_orders

完整脚本如下：

cat > /tmp/hudi_write_ops_demo.py <<'PY'
from pyspark.sql import SparkSession

spark = (SparkSession.builder
    .appName("hudi-write-ops-demo")
    .master("local[2]")
    .config("spark.sql.catalogImplementation", "in-memory")
    .getOrCreate())

base_path = "hdfs://localhost:9000/hudi/demo_write_ops_orders"
schema = ["order_id", "dt", "status", "amount", "ts"]

common = {
    "hoodie.table.name": "demo_write_ops_orders",
    "hoodie.datasource.write.table.name": "demo_write_ops_orders",
    "hoodie.datasource.write.table.type": "COPY_ON_WRITE",
    "hoodie.datasource.write.recordkey.field": "order_id",
    "hoodie.datasource.write.partitionpath.field": "dt",
    "hoodie.datasource.write.precombine.field": "ts",
    "hoodie.datasource.write.hive_style_partitioning": "true",
    "hoodie.metadata.enable": "true",
}

def write_rows(rows, operation, mode="append", extra=None):
    df = spark.createDataFrame(rows, schema)
    writer = df.write.format("hudi")
    opts = dict(common)
    opts["hoodie.datasource.write.operation"] = operation
    if extra:
        opts.update(extra)
    for k, v in opts.items():
        writer = writer.option(k, v)
    writer.mode(mode).save(base_path)

def show_table(title):
    print(title)
    (spark.read.format("hudi")
        .load(base_path)
        .select("_hoodie_commit_time", "order_id", "dt", "status", "amount", "ts")
        .orderBy("order_id")
        .show(truncate=False))

bulk_rows = [
    ("3001", "2025-07-28", "CREATED", 100.0, "20250728083000"),
    ("3002", "2025-07-28", "CREATED", 30.0, "20250728083010"),
    ("3003", "2025-07-29", "CREATED", 60.0, "20250728083020"),
]

insert_rows = [
    ("3004", "2025-07-29", "CREATED", 18.5, "20250728083500"),
]

upsert_rows = [
    ("3001", "2025-07-28", "PAID", 100.0, "20250728083800"),
    ("3005", "2025-07-29", "CREATED", 45.0, "20250728083810"),
]

delete_rows = [
    ("3002", "2025-07-28", "DELETE", 0.0, "20250728083900"),
]

overwrite_rows = [
    ("3999", "2025-07-29", "RECALCULATED", 999.0, "20250728084000"),
]

write_rows(bulk_rows, "bulk_insert", mode="overwrite")
show_table("=== after bulk_insert ===")

write_rows(insert_rows, "insert")
show_table("=== after insert ===")

write_rows(upsert_rows, "upsert")
show_table("=== after upsert ===")

write_rows(delete_rows, "delete")
show_table("=== after delete order_id=3002 ===")

write_rows(overwrite_rows, "insert_overwrite")
show_table("=== after insert_overwrite dt=2025-07-29 ===")

print("=== distinct commits ===")
(spark.read.format("hudi")
    .load(base_path)
    .select("_hoodie_commit_time")
    .distinct()
    .orderBy("_hoodie_commit_time")
    .show(truncate=False))

spark.stop()
PY

执行命令：

source /etc/profile.d/bigdata_env.sh

hdfs dfs -rm -r -f /hudi/demo_write_ops_orders

spark-submit \
  --master local[2] \
  --conf spark.sql.catalogImplementation=in-memory \
  --jars /opt/module/hudi/lib/hudi-spark3.5-bundle_2.12-1.0.2.jar \
  /tmp/hudi_write_ops_demo.py

后续 hdfs dfs 命令中的 /hudi/demo_write_ops_orders 与 Spark 代码中的 hdfs://localhost:9000/hudi/demo_write_ops_orders 指向同一张 Hudi 表，只是省略了默认文件系统前缀。本文示例使用 --jars 显式指定 Hudi Spark bundle；如果已经在 Spark jars 目录中放入同版本 bundle，可以省略 --jars，但不要同时混入多个 Hudi 版本的 bundle。

这里的 mode="overwrite" 只用于第一次初始化创建测试表；后续持续写入统一使用 append，并通过 hoodie.datasource.write.operation 表达 Hudi 写入语义。Spark 的 save mode 控制 DataFrameWriter 对目标路径的写入模式，Hudi 的 operation 控制表语义，二者不要混为一谈。

2.4 写入结果验证

bulk_insert 后：

=== after bulk_insert ===
+-------------------+--------+----------+-------+------+--------------+
|_hoodie_commit_time|order_id|dt        |status |amount|ts            |
+-------------------+--------+----------+-------+------+--------------+
|20250729084215109  |3001    |2025-07-28|CREATED|100.0 |20250728083000|
|20250729084215109  |3002    |2025-07-28|CREATED|30.0  |20250728083010|
|20250729084215109  |3003    |2025-07-29|CREATED|60.0  |20250728083020|
+-------------------+--------+----------+-------+------+--------------+

insert 后：

=== after insert ===
+-------------------+--------+----------+-------+------+--------------+
|_hoodie_commit_time|order_id|dt        |status |amount|ts            |
+-------------------+--------+----------+-------+------+--------------+
|20250729084215109  |3001    |2025-07-28|CREATED|100.0 |20250728083000|
|20250729084215109  |3002    |2025-07-28|CREATED|30.0  |20250728083010|
|20250729084215109  |3003    |2025-07-29|CREATED|60.0  |20250728083020|
|20250729084231194  |3004    |2025-07-29|CREATED|18.5  |20250728083500|
+-------------------+--------+----------+-------+------+--------------+

upsert 后：

=== after upsert ===
+-------------------+--------+----------+-------+------+--------------+
|_hoodie_commit_time|order_id|dt        |status |amount|ts            |
+-------------------+--------+----------+-------+------+--------------+
|20250729084235344  |3001    |2025-07-28|PAID   |100.0 |20250728083800|
|20250729084215109  |3002    |2025-07-28|CREATED|30.0  |20250728083010|
|20250729084215109  |3003    |2025-07-29|CREATED|60.0  |20250728083020|
|20250729084231194  |3004    |2025-07-29|CREATED|18.5  |20250728083500|
|20250729084235344  |3005    |2025-07-29|CREATED|45.0  |20250728083810|
+-------------------+--------+----------+-------+------+--------------+

这里能看到 3001 被更新成 PAID，3005 被新增。

delete 后：

=== after delete order_id=3002 ===
+-------------------+--------+----------+-------+------+--------------+
|_hoodie_commit_time|order_id|dt        |status |amount|ts            |
+-------------------+--------+----------+-------+------+--------------+
|20250729084235344  |3001    |2025-07-28|PAID   |100.0 |20250728083800|
|20250729084215109  |3003    |2025-07-29|CREATED|60.0  |20250728083020|
|20250729084231194  |3004    |2025-07-29|CREATED|18.5  |20250728083500|
|20250729084235344  |3005    |2025-07-29|CREATED|45.0  |20250728083810|
+-------------------+--------+----------+-------+------+--------------+

3002 已经不在最新快照中。

insert_overwrite 覆盖 dt=2025-07-29 后：

=== after insert_overwrite dt=2025-07-29 ===
+-------------------+--------+----------+------------+------+--------------+
|_hoodie_commit_time|order_id|dt        |status      |amount|ts            |
+-------------------+--------+----------+------------+------+--------------+
|20250729084235344  |3001    |2025-07-28|PAID        |100.0 |20250728083800|
|20250729084242313  |3999    |2025-07-29|RECALCULATED|999.0 |20250728084000|
+-------------------+--------+----------+------------+------+--------------+

这个结果最关键：dt=2025-07-29 下面原来的 3003、3004、3005 都被新的分区视图替换，只剩 3999。但 dt=2025-07-28 的 3001 还保留。

2.5 Timeline 验证

查看 Timeline：

hdfs dfs -ls /hudi/demo_write_ops_orders/.hoodie/timeline

输出节选如下：

-rw-r--r--   1 root supergroup       4611 2025-07-29 08:42 /hudi/demo_write_ops_orders/.hoodie/timeline/20250729084215109_20250729084227905.commit
-rw-r--r--   1 root supergroup       4060 2025-07-29 08:42 /hudi/demo_write_ops_orders/.hoodie/timeline/20250729084231194_20250729084234398.commit
-rw-r--r--   1 root supergroup       4360 2025-07-29 08:42 /hudi/demo_write_ops_orders/.hoodie/timeline/20250729084235344_20250729084238804.commit
-rw-r--r--   1 root supergroup       4062 2025-07-29 08:42 /hudi/demo_write_ops_orders/.hoodie/timeline/20250729084239354_20250729084241420.commit
-rw-r--r--   1 root supergroup       4223 2025-07-29 08:42 /hudi/demo_write_ops_orders/.hoodie/timeline/20250729084242313_20250729084244978.replacecommit

在 Hudi 1.0.x 中，完成态 Instant 文件名里可以看到 begin instant 和 completion instant，例如 begin_completion.commit；这里前半段更接近写入请求时间，后半段是完成时间。

前几次写入都是 commit，其中 delete 也表现为一次 commit，因为它通过新的提交改变快照可见性。最后一次 insert_overwrite 变成了 replacecommit，因为它替换的是分区范围内的一组 FileGroup。这说明分区覆盖是 Hudi 的表级替换语义，而不是简单手动删除 HDFS 目录。

最终最新快照中的记录只带有两个 _hoodie_commit_time：

=== distinct commits ===
+-------------------+
|_hoodie_commit_time|
+-------------------+
|20250729084235344  |
|20250729084242313  |
+-------------------+

这里 3001 来自 upsert 提交，3999 来自 insert_overwrite 提交。历史提交仍然存在于 Timeline 中，只是它们产生的部分文件视图已经被后续 commit 或 replacecommit 替换，不再出现在当前 Snapshot Query 结果里。

3. 常见问题与选型

3.1 常见问题

3.1.1 `insert` 和 `upsert` 怎么选

如果输入数据一定是新增，且上游已经保证不会重复发送同一个 record key，可以选 insert。只要存在重放、补偿、重复事件或状态更新，就应优先考虑 upsert 或额外去重策略。

不要为了快把 CDC 数据写成 insert。这会把更新事件变成追加事件，下游查询最新状态时很容易出错。

3.1.2 `bulk_insert` 能不能长期用于增量

不建议把 bulk_insert 当成日常 CDC 更新写入方式。它更适合初始化导入或特殊的大批量追加场景；持续增量要先看业务语义：纯追加用 insert，有更新用 upsert。

3.1.3 删除是物理删除吗

不是立刻物理删除。Delete 先通过新的提交改变表快照中的可见性。旧文件什么时候被物理清理，取决于 clean 策略。

3.1.4 `insert_overwrite` 会不会影响其他分区

在配置正确、输入数据只包含目标分区、并且使用的是分区级 insert_overwrite 语义时，只应该影响目标分区。执行前最好先确认输入 DataFrame 的分区值，避免误覆盖更多分区。本文示例覆盖 dt=2025-07-29 后，dt=2025-07-28 的 3001 仍然存在。

3.1.5 为什么 Timeline 里有 `replacecommit`

因为 insert_overwrite 属于替换旧文件组的写入语义。Hudi 需要在 Timeline 中记录哪些旧 FileGroup 被替换，哪些新 FileGroup 成为当前视图。replacecommit 不只可能出现在 insert_overwrite 场景，后续 clustering 等替换 FileGroup 的表服务也可能产生类似替换语义；本文只关注 insert_overwrite。

3.1.6 为什么最终只看到两个 commit time

Snapshot Query 只看当前有效文件视图。历史 commit 对应的文件可能还在 HDFS 上，但如果已经被后续 commit 或 replacecommit 替换，就不会出现在最新查询结果里。

3.1.7 为什么 HDFS 上还有旧文件，查询却看不到

这在 insert_overwrite、COW 更新和 clean 策略里都可能出现。

Hudi 不会把文件系统目录当成唯一事实来源。查询端会根据 Timeline 和 FileSlice 构造当前文件视图。旧文件可能还在 HDFS 上，但它们可能已经被新的 commit 或 replacecommit 替换，只是还没到 cleaner 物理删除的时间。

所以排查时要同时看三件事：

hdfs dfs -ls /hudi/demo_write_ops_orders/.hoodie/timeline
hdfs dfs -ls /hudi/demo_write_ops_orders/dt=2025-07-29
spark.read.format("hudi").load("hdfs://localhost:9000/hudi/demo_write_ops_orders").show(false)

目录里“看得到文件”不等于 Snapshot Query 会选择它。

3.2 写入操作选型

Hudi 写入操作最容易误解的地方，是把它们都看成“保存一批 DataFrame”。

实际上，bulk_insert、insert、upsert、delete、insert_overwrite 背后的表语义完全不同。bulk_insert 偏初始化，insert 偏追加，upsert 偏变更，delete 改变可见性，insert_overwrite 替换分区或整表视图。

我更愿意把写操作和业务场景绑定起来：

初始化历史数据：bulk_insert。
日志事件追加：insert。
CDC 状态同步：upsert。
删除事件：delete 或 _hoodie_is_deleted。
分区重算：insert_overwrite。

这样选比单纯记参数更可靠。因为 Hudi 的复杂度不只在 API，而在表语义、索引成本、文件版本和 Timeline 可见性。

4. 总结

这篇主要讲了 Hudi 的几类写操作。

upsert 会判断 insert / update，适合 CDC。
insert 适合纯追加，通常比 upsert 少了更新定位成本，但不适合更新语义。
bulk_insert 适合初始化大批量导入。
delete 通过新的提交让记录在快照中不可见，不是立刻物理删除旧文件。
insert_overwrite 用 replacecommit 覆盖分区或表视图，适合分区修复和重算。
record key、partition path、precombine field 是写入语义的基础。
Key Generator 要同时考虑多字段 record key、多字段分区、Hive style path 和分区字段稳定性。
Hudi 写入结果一定要结合查询结果、HDFS 文件和 Timeline 一起看。

下一篇会进入 Spark 集成 Hudi：DataSource、Spark SQL 和代码实战。到那时会把本篇的写操作放到 Spark Shell、Spark SQL 和 spark-submit 中进一步展开。

hnbian

https://www.hnbian.cn/posts/c8f4d6a2.html

本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 hnbian !

spark hudi write upsert insert delete

Apache Hudi 学习笔记 07：Spark 集成 Hudi 实战

2025-08-05 hudi

spark hudi data-lake lakehouse spark-sql

Apache Hudi 学习笔记 05：COW 与 MOR 表类型，以及三种查询方式

2025-07-17 hudi

hudi data-lake cow mor query compaction

Apache Hudi 学习笔记 06：Hudi 写入流程与写操作类型详解

1. 写入语义和操作类型

1.1 背景问题

1.2 写入前的几个关键概念

1.3 UPSERT

1.4 INSERT

1.5 BULK_INSERT

1.6 DELETE

1.7 INSERT_OVERWRITE

2. 写入策略和实测验证

2.1 Key 生成策略

2.2 Small File 策略

2.3 Spark 示例：连续执行多种写操作

2.4 写入结果验证

2.5 Timeline 验证

3. 常见问题与选型

3.1 常见问题

3.1.1 insert 和 upsert 怎么选

3.1.2 bulk_insert 能不能长期用于增量

3.1.3 删除是物理删除吗

3.1.4 insert_overwrite 会不会影响其他分区

3.1.5 为什么 Timeline 里有 replacecommit

3.1.6 为什么最终只看到两个 commit time

3.1.7 为什么 HDFS 上还有旧文件，查询却看不到

3.2 写入操作选型

4. 总结

3.1.1 `insert` 和 `upsert` 怎么选

3.1.2 `bulk_insert` 能不能长期用于增量

3.1.4 `insert_overwrite` 会不会影响其他分区

3.1.5 为什么 Timeline 里有 `replacecommit`