企业官网建设流程全解析

一、引言

在数据湖场景中，写入操作并非简单的"append"，不同业务场景对数据写入有截然不同的需求：

• CDC 增量同步：需要高效的 upsert 能力，同时还需处理上游的 DELETE 事件
• 日志类数据：追求高吞吐的纯追加写入
• 历史数据回填：需要大批量的初始加载
• 分区级数据重刷：需要原子性的覆盖写入
• 数据合规删除（GDPR/CCPA）：需要精准定位并物理移除记录

Hudi 通过提供多种 Write Operation 类型，让用户根据场景精确控制写入行为，从而在写入性能、数据一致性、存储效率之间取得最优平衡。

二、Hudi Write Operations 详解

Hudi 提供以下核心 Write Operation 类型与定位：

1. UPSERT

UPSERT 是 Hudi 最核心的写操作，也是 COW 和 MOR 表的默认写操作。其核心逻辑是：先通过索引定位记录是否已存在，存在则更新，不存在则插入。

COW 表的 UPSERT：

• 更新操作会重写整个文件（Copy-On-Write），即读取原始 Parquet 文件，合并更新记录后写出新版本文件。
• 写入放大较高，但读取时无需合并，查询性能最优。

MOR 表的 UPSERT：

• 更新操作写入 Delta Log 文件，不立即重写 Base 文件。
• 写入效率高，但读取时需要合并 Base + Log。
• 通过后续 Compaction 操作将 Log 合入 Base 文件。

关键配置：

# 写操作类型 hoodie.datasource.write.operation=upsert # Payload 合并策略 hoodie.datasource.write.payload.class=org.apache.hudi.common.model.OverwriteWithLatestAvroPayload # precombine 字段（必须指定，用于同 key 去重） hoodie.datasource.write.precombine.field=ts # 并行度控制 hoodie.upsert.shuffle.parallelism=200

2. INSERT

INSERT 操作跳过索引查找步骤，直接将数据写入新文件或追加到现有小文件中（小文件合并策略）。

关键特点：

• 不执行索引查找，因此写入性能高于 UPSERT
• 如果数据中存在与已有数据重复的 key，不会去重（除非显式开启 hoodie.combine.before.insert，该配置仅做批次内去重）
• 适用于能从业务层保证无重复的场景

与 UPSERT 性能对比：

性能（吞吐量）: BULK_INSERT > INSERT > UPSERT 数据正确性保障: UPSERT > INSERT > BULK_INSERT

3. BULK_INSERT

BULK_INSERT 专为大批量初始数据加载设计，绕过了索引查找和小文件分配逻辑，直接利用 Spark/Flink 的排序能力对数据进行组织后写入。

排序模式（hoodie.bulkinsert.sort.mode）：

关键配置：

hoodie.datasource.write.operation=bulk_insert hoodie.bulkinsert.sort.mode=PARTITION_SORT hoodie.bulkinsert.shuffle.parallelism=300

4. DELETE

Hudi 的删除并非单一机制，而是提供了一组完整的删除能力体系，涵盖软删除和硬删除两大类。

1.软删除

软删除保留记录在存储中的存在，但将所有非主键字段（除 record key、partition path 和 precombine field 外）置为 null。

# 软删除示例：保留 key 字段，其余置空 soft_delete_df = spark.createDataFrame( [(record_key, partition_path, precombine_val, None, None, ...)], schema ) soft_delete_df.write.format("hudi") \ .option("hoodie.datasource.write.operation", "upsert") \ .option("hoodie.table.name", "my_table") \ .mode("append") \ .save(path)

适用场景：需要在增量查询（Incremental Query）中感知到"删除事件"的下游消费者。

2.硬删除

硬删除会从存储中彻底移除记录（经 Compaction 和 Clean 后物理删除）。

• DELETE Operation

最直接的方式，传入的 DataFrame 只需包含 record key 和 partition path：

# 硬删除 - DELETE operation delete_df = spark.createDataFrame( [(record_key, partition_path)], ["id", "partition"] ) delete_df.write.format("hudi") \ .option("hoodie.datasource.write.operation", "delete") \ .option("hoodie.datasource.write.recordkey.field", "id") \ .option("hoodie.datasource.write.partitionpath.field", "partition") \ .option("hoodie.table.name", "my_table") \ .mode("append") \ .save(path)

• EmptyHoodieRecordPayload

通过指定空 payload，使得 UPSERT 写入时将匹配记录标记为删除：

hoodie.datasource.write.operation=upsert hoodie.datasource.write.payload.class=org.apache.hudi.common.model.EmptyHoodieRecordPayload

• _hoodie_is_deleted字段

在 CDC 管道中最为常用，可以将 insert/update/delete 事件统一在一个 UPSERT 流中处理：

from pyspark.sql.functions import lit, when, col # CDC 事件流统一处理 cdc_df = cdc_df.withColumn( "_hoodie_is_deleted", when(col("op") == "d", lit(True)).otherwise(lit(False)) ) cdc_df.write.format("hudi") \ .option("hoodie.datasource.write.operation", "upsert") \ .option("hoodie.table.name", "my_table") \ .mode("append") \ .save(path)

3.不同表类型的删除行为与选型建议

不同表类型下的删除行为：

删除方式选型建议：

5. INSERT_OVERWRITE

NSERT_OVERWRITE 以分区级别进行原子替换，它为目标分区创建全新的 FileGroup，然后在 Timeline 上原子性地将旧文件标记为替换。

核心优势：

• 原子性：要么全部成功，要么全部回滚
• 不影响未涉及的分区
• 无需索引查找，性能好
• 自动清理旧数据（通过 Cleaner）
• 可用于分区级删除（写入空 DataFrame）

关键配置：

# Spark 示例：分区级数据重刷 df.write.format("hudi") \ .option("hoodie.datasource.write.operation", "insert_overwrite") \ .option("hoodie.datasource.write.partitionpath.field", "dt") \ .option("hoodie.datasource.write.recordkey.field", "id") \ .option("hoodie.table.name", "my_table") \ .mode("append") \ .save("s3://bucket/my_table")

6. INSERT_OVERWRITE_TABLE

与 INSERT_OVERWRITE 类似，但作用范围是全表。整表数据被新写入的数据完全替换。

适用于全量刷新场景，如每日全量快照表、维度表全量更新。

三、版本差异与选型建议

相较于 0.x 版本，Hudi 1.x 在写入路径上有以下关键演进：

Hudi写操作场景选型决策指南：

四、最佳实践

• UPSERT 性能优化

# 1. 索引选型 - 根据数据规模选择 # 小规模（< 10亿记录）：Bloom Index hoodie.index.type=BLOOM hoodie.bloom.index.filter.type=DYNAMIC_V0 # 大规模 + 高频更新：Bucket Index（Hudi 1.x 推荐） hoodie.index.type=BUCKET hoodie.index.bucket.engine=CONSISTENT_HASHING hoodie.bucket.index.num.buckets=256 # 2. 并行度调优 hoodie.upsert.shuffle.parallelism=<num_partitions * 2~3> # 3. precombine field 必须指定 hoodie.datasource.write.precombine.field=updated_at

• BULK_INSERT 初始加载

# 建议配置 hoodie.datasource.write.operation=bulk_insert hoodie.bulkinsert.sort.mode=PARTITION_SORT # 控制文件大小（避免过多小文件） hoodie.parquet.max.file.size=134217728 # 128MB hoodie.parquet.small.file.limit=0 # 关闭小文件合并 # 并行度 = 数据总量 / 目标文件大小 hoodie.bulkinsert.shuffle.parallelism=<total_data_size / 128MB>

• MOR 表 + UPSERT 的 Compaction 策略

# Compaction 触发策略 hoodie.compact.inline=false # 生产环境关闭 inline compaction hoodie.compact.schedule.inline=true # 在写入时调度 compaction plan hoodie.compact.inline.max.delta.commits=5 # 每5次提交触发一次 # 异步 Compaction（推荐） # 通过独立作业执行 compaction，避免影响写入延迟