Apache Hudi 学习笔记 09：并发控制、锁机制与 Spark 写入调优

前面几篇已经把 Hudi 的表类型、查询方式、写入流程、Spark 集成和 Kafka 导入都跑了一遍。到这里其实已经可以用 Hudi 做不少测试任务了，但如果要把它放进稍微真实一点的链路里，还会遇到两个问题：多个写任务能不能同时写一张表？Spark 写入慢、文件多、内存抖动时应该从哪里调？

这两个问题看起来一个偏事务，一个偏性能，但在 Hudi 里经常会碰到一起。比如一个 Spark 任务在做离线修复分区，另一个任务在跑近实时 upsert；再比如写入并行度调大以后，Timeline 提交、索引定位、小文件合并和 executor 内存都会一起变化。

1. 并发控制基础

1.1 背景问题

普通 Hive 外表或 Parquet 目录通常没有完整的表级事务语义。一个任务写文件，另一个任务读目录，中间如果出现失败、覆盖、清理、重复提交，很容易出现下面这些问题：

1. 读任务读到一半新旧文件混在一起。
2. 两个写任务同时改同一个分区，后写入任务覆盖前一个任务结果。
3. 清理任务删掉旧文件，但某个读任务还在依赖旧快照。
4. 更新类写入需要定位老记录，索引和 shuffle 成本非常高。
5. 小文件越来越多，查询慢，写入也越来越慢。

Hudi 解决这些问题的核心不是“加一个锁”这么简单，而是把 Timeline、Instant、FileGroup、FileSlice、索引、清理服务、压缩服务组合成一套表管理机制。并发控制负责保证提交语义，Spark 调优负责让写入成本可控。

1.2 Hudi 并发控制解决什么

Hudi 官方文档把并发控制放在 writers、readers、table services 之间讨论，这一点很重要。实际生产里不只有写任务，还有 compaction、clean、clustering、Hive Sync、查询任务。它们都可能在同一张 Hudi 表上同时工作。

我把 Hudi 的并发关系简单分成三类：

并发关系	典型场景	主要机制
writer 和 reader	Spark 正在写，Hive / Spark 正在查	MVCC、Snapshot Isolation
writer 和 table service	写入任务和 clean / compaction 同时运行	MVCC、Timeline 协调
writer 和 writer	多个 Spark / Flink 任务同时写一张表	Optimistic Concurrency Control、锁服务

如果 table services 在同一个 writer 进程中 inline 或 async 执行，通常不需要额外分布式锁；如果 compaction、clustering、clean 等被拆成独立作业，就要按多写入 / 多服务并发模型重新评估锁和冲突处理。

默认情况下，很多测试环境其实是 single writer：同一时刻只有一个任务写同一张表。这种模型最简单，也最稳定。如果要启用多 writer，就要额外考虑锁服务和文件级冲突判断。

1.3 MVCC 与快照隔离

Hudi 的 MVCC 可以理解成“新提交生成新版本，旧版本不会立刻消失”。读任务只读取已经完成的 Instant 对应的文件视图，不会去读正在写入中的中间状态。

这张图主要说明 reader 通过 Timeline 读取 completed instant 对应的 FileSlice，因此不会读到 writer 正在写入的中间状态。

这个机制解决的是读写并发问题：

1. Writer 写入时先产生新的 FileSlice 或新 Base File。
2. Timeline 中的 Instant 会经历 requested、inflight、completed 等状态。
3. 普通 Snapshot Query 构造可见文件视图时，主要基于已经完成的写入类 Instant，同时也会受到已完成的 rollback、clean、replacecommit、compaction、clustering 等 Timeline action 影响。
4. 新 commit 完成后，后续查询才会切到新的快照。
5. clean 服务会根据保留策略清理旧版本，而不是写完马上删除。

生产环境中需要确保 cleaner 保留策略覆盖长查询、增量消费和故障恢复所需的历史版本，否则旧查询或落后的增量任务可能找不到需要的旧文件。

所以 Hudi 的读写并发不是靠读写互斥锁来保证的。读任务不需要阻塞写任务，写任务也不会原地覆盖读任务正在使用的文件。这个模型非常适合数据湖，因为 HDFS、对象存储这类系统本来就更适合追加新文件，而不是频繁原地更新。

1.4 OCC 与多 writer

如果两个写任务同时写同一张表，仅有 MVCC 还不够。因为两个 writer 可能都基于同一个旧快照开始写，最后同时尝试提交新 Instant。Hudi 的 Optimistic Concurrency Control 处理的就是这类 writer 和 writer 之间的并发。

这张图主要说明锁服务只负责提交阶段的协调，而 OCC 的核心仍然是写入冲突检测和 Timeline 提交语义。

OCC 的关键点有两个：

1. 写任务先乐观地执行，不是一开始就把整张表锁住。
2. 提交前会做文件级冲突判断；如果两个 writer 的写入涉及重叠文件或相同 FileGroup，就可能发生冲突。

也就是说，锁服务负责协调提交阶段的临界区，但真正判断能不能提交的是文件级冲突检测。在正确配置 OCC 和外部锁服务的前提下，两个 writer 如果写的是不同分区、不同 FileGroup，理论上可以都成功；如果写入范围重叠，则可能触发冲突检测并让其中一个提交失败。

官方多 writer 的核心配置方向如下，真实使用时还必须补齐具体 lock provider 的参数：

hoodie.write.concurrency.mode=optimistic_concurrency_control
hoodie.cleaner.policy.failed.writes=LAZY
hoodie.write.lock.provider=org.apache.hudi.client.transaction.lock.ZookeeperBasedLockProvider
hoodie.write.lock.zookeeper.url=localhost
hoodie.write.lock.zookeeper.port=2181
hoodie.write.lock.zookeeper.base_path=/hudi/locks

这里要注意 hoodie.cleaner.policy.failed.writes=LAZY。多 writer 场景下，失败写入的清理不能过于激进，否则可能影响其他并发写入或恢复流程。

1.5 锁服务怎么选

Hudi 1.0.2 支持多种锁实现，常见选择如下：

锁实现	适用场景	判断
InProcessLockProvider	单 JVM / 单 writer + 同进程 table service	只适合同一 JVM / 单 writer + 同进程 table service 场景，不是分布式多 writer 锁
ZookeeperBasedLockProvider	Hadoop / Spark 集群常见	依赖明确，适合已有 ZooKeeper 的环境
HiveMetastoreBasedLockProvider	已经有 Hive Metastore	依赖 HMS，可减少新组件
DynamoDBBasedLockProvider	AWS 环境	适合云上 S3 数据湖
StorageBasedLockProvider	S3 条件写锁场景	Hudi 1.0.2 文档中主要对应 S3 条件写锁能力，不应泛化成所有对象存储都可用
FileSystemBasedLockProvider	本地 / 文件系统测试	只适合本地 / 文件系统测试，不建议生产多 writer 使用

如果只是 Spark 单任务写入 Hudi 表，我会优先保持 single writer，不主动打开 OCC。single writer 并不代表不能有 reader；读写并发仍由 Timeline 和 MVCC 保证。多 writer 不只是多写几个参数，还要保证锁服务稳定、失败写入可清理、不同任务写入范围可控。

如果确实有多 writer，比如一个任务负责实时 upsert，一个任务负责局部分区修复，那就要把写入范围设计清楚。最怕的是两个任务都随意写全表，最后冲突频繁，吞吐反而更差。

1.5.1 Spark DataFrame 和 HoodieStreamer 怎么开启 OCC

如果用 Spark DataFrame 写入 Hudi，可以把 OCC 参数直接放进 DataSource options。下面只是 ZooKeeper 锁的参数形态示例，只有在已经单独部署 ZooKeeper 的环境中才可使用；本文前面 Kafka 的 KRaft 模式并不自带 ZooKeeper。这段代码是多 writer 配置示例，不是本文第 7 节单机调优脚本的必执行步骤：

concurrency_options = {
    "hoodie.write.concurrency.mode": "optimistic_concurrency_control",
    "hoodie.cleaner.policy.failed.writes": "LAZY",
    "hoodie.write.lock.provider": "org.apache.hudi.client.transaction.lock.ZookeeperBasedLockProvider",
    "hoodie.write.lock.zookeeper.url": "localhost",
    "hoodie.write.lock.zookeeper.port": "2181",
    "hoodie.write.lock.zookeeper.base_path": "/hudi/locks"
}

(
    df.write.format("hudi")
    .options(**common_options)
    .options(**concurrency_options)
    .mode("append")
    .save(base_path)
)

如果用前一篇的 HoodieStreamer，也是在 props 里加同样的并发参数。把这些 OCC 参数追加到同一个 HoodieStreamer props 文件中即可，但要确认不要同时混用多个 lock provider 或不同 concurrency.mode：

hoodie.write.concurrency.mode=optimistic_concurrency_control
hoodie.cleaner.policy.failed.writes=LAZY
hoodie.write.lock.provider=org.apache.hudi.client.transaction.lock.ZookeeperBasedLockProvider
hoodie.write.lock.zookeeper.url=localhost
hoodie.write.lock.zookeeper.port=2181
hoodie.write.lock.zookeeper.base_path=/hudi/locks

然后继续使用 HoodieStreamer 提交：

CP=/opt/module/hudi-1.0.2/lib/hudi-utilities-bundle_2.12-1.0.2.jar:/opt/module/hudi-1.0.2/lib/hudi-spark3.5-bundle_2.12-1.0.2.jar

spark-submit \
  --master local[2] \
  --conf spark.driver.extraClassPath=$CP \
  --conf spark.executor.extraClassPath=$CP \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --conf spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar \
  --class org.apache.hudi.utilities.streamer.HoodieStreamer \
  /opt/module/hudi-1.0.2/lib/hudi-utilities-bundle_2.12-1.0.2.jar \
  --props file:///opt/module/hudi-1.0.2/streamer-conf/kafka-source.properties \
  --schemaprovider-class org.apache.hudi.utilities.schema.FilebasedSchemaProvider \
  --source-class org.apache.hudi.utilities.sources.JsonKafkaSource \
  --source-ordering-field ts \
  --target-base-path /hudi/demo_deltastreamer_orders \
  --target-table demo_deltastreamer_orders \
  --table-type COPY_ON_WRITE \
  --op UPSERT \
  --source-limit 1000

前面第 2 篇的 Kafka 使用 KRaft 模式，不依赖 ZooKeeper。所以下面的 ZooKeeper 锁示例有一个前提：集群已经单独部署了 ZooKeeper，或者你明确决定为了 Hudi 多 writer 引入 ZooKeeper。如果没有 ZooKeeper，也可以优先考虑 Hive Metastore 锁。

如果要确认 ZooKeeper 锁路径是否真的被使用，可以在任务运行时打开客户端查看：

zkCli.sh -server localhost:2181

ls /hudi/locks

本文前几篇没有单独安装 ZooKeeper；如果要使用 ZK 锁，需要先安装 ZooKeeper 客户端和服务端，zkCli.sh 只是验证锁路径的示例命令。如果从旧文章迁移配置，可能会看到更多 ZK lock 相关参数；本文先保留 Hudi 1.0.2 文档中的基础必填项，其他参数应按当前配置页核对后再加。

2. Spark 写入调优实践

2.1 Spark 写入调优先看哪几个点

Spark 写 Hudi 慢，通常不是一个参数的问题。我的排查顺序一般是：

1. 数据规模和输入分区是否合理。
2. hoodie.upsert.shuffle.parallelism 等 shuffle 并行度是否过低。
3. 索引类型是否适合当前表。
4. 文件大小策略是否导致大量小文件。
5. executor 内存、off-heap、GC、spill 是否异常。
6. 是否存在跨分区更新、全局索引或极端数据倾斜。

下面这张图是我对 Spark 写入 Hudi 调优链路的理解：

这张图主要说明 Spark 写入 Hudi 的性能问题通常同时涉及输入分区、shuffle、索引、小文件、内存和 Timeline 提交，不应只改一个参数。

官方调优建议里有一个很实用的经验值：hoodie.[insert|upsert|bulkinsert].shuffle.parallelism 可以按 input_data_size / 500MB 起步，但这只是初始估算；最终还要结合 executor 数、cores、目标文件大小、分区数量和 Spark UI 观察结果调整。这个值不是绝对标准，但能避免一个常见问题：输入数据已经很大，写入并行度还停留在默认小值，导致少数 task 承担大量写入和索引定位工作。

2.1.1 常规调优参数拆解

常规调优建议可以整理成下面几类。这里不机械照抄旧参数，而是按 Spark 3.5 和 Java 11 的环境重新组织。

调优点	关注问题	常用参数
shuffle 并行度	少数 task 写入过重，索引定位慢	hoodie.upsert.shuffle.parallelism、hoodie.insert.shuffle.parallelism、hoodie.bulkinsert.shuffle.parallelism
off-heap / overhead	Parquet 写入、Netty、压缩、容器额外内存	spark.executor.memoryOverhead、spark.driver.memoryOverhead
Spark 内存	merge、compaction、缓存输入数据	spark.memory.fraction、spark.memory.storageFraction
文件大小	小文件过多或单文件过大	hoodie.parquet.max.file.size、hoodie.parquet.small.file.limit
BloomFilter	大表索引定位慢、假阳性过多	hoodie.bloom.index.num_entries、hoodie.bloom.index.fpp
GC / OOM	executor 长时间 GC 或直接 OOM	spark.executor.extraJavaOptions、heap dump、Spark UI

off-heap 或容器 overhead 可以先这样设置：

spark.executor.memory=6g
spark.executor.memoryOverhead=3g
spark.driver.memory=4g
spark.driver.memoryOverhead=1g

这里的 memoryOverhead 主要是给 JVM 之外的额外内存、Netty、压缩、Parquet 写入和容器开销预留空间，不等同于一定启用 Spark off-heap execution memory。是否启用 spark.memory.offHeap.enabled 需要单独评估。

如果是 YARN 部署，旧版示例里常见 spark.yarn.executor.memoryOverhead 和 spark.yarn.driver.memoryOverhead。在 Spark 3.x 中我更倾向于先使用通用的 spark.executor.memoryOverhead 和 spark.driver.memoryOverhead，再按集群部署模式补充 YARN 专属配置。

Spark memory 参数不要随便调小或调大。遇到 Hudi merge 或 compaction OOM 时，可以先从下面的保守值开始试：

spark.memory.fraction=0.4
spark.memory.storageFraction=0.2

这组值只是 OOM 排查时的保守尝试，不是默认推荐值。修改前后要对比 Spark UI 中 spill、GC time、stage duration 和失败率。这类参数的目标不是“让所有数据都放进内存”，而是在内存不够时尽量让 Spark spill，避免 executor 直接崩掉。速度变慢通常比任务反复 OOM 更容易接受。

对于时间序列、事件日志、数据库 changelog 这类数据，如果单分区记录量很大，Bloom Index 参数也值得关注：

hoodie.index.type=BLOOM
hoodie.bloom.index.num_entries=60000
hoodie.bloom.index.fpp=0.0000001

num_entries 估计的是单个文件里 record key 数量，fpp 控制假阳性概率。它们不是越大越好，BloomFilter 越精细，元数据和判断成本也会增加。不要为了追求极低假阳性率盲目把 fpp 调到特别小。更重要的是 key 设计：如果业务允许，partition path 或 record key 的设计可以体现时间 / 区域等范围特征，帮助剪枝；但 record key 必须保持同一业务记录的稳定性，不能为了剪枝把会变化的时间字段拼进去导致重复 key。

Java 11 下 GC 日志建议使用统一日志参数，而不是旧的 CMS 示例：

spark.executor.extraJavaOptions=-XX:+UseG1GC -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/tmp/hoodie-heapdump.hprof -Xlog:gc*:file=/tmp/spark-executor-gc-%p.log:time,uptime,level,tags:filecount=5,filesize=100M

生产环境需要确认 executor 节点对该路径有写权限，并考虑日志滚动、收集和清理；否则 GC 日志和 heap dump 可能占满本地磁盘。

如果集群还在 Java 8，可以按实际情况使用 CMS 或 G1，但不要把旧版 GC 参数原样复制到 Java 11+ 环境里。

2.2 写入调优参数验证表

下面用 Spark 写一张 COW 表，主要验证调优参数的配置方式和文件结果，不做性能基准对比；真正调优需要结合 Spark UI、任务耗时、spill、GC 和文件数量多轮对比。本文第 7 节不是多 writer 并发冲突测试，只是 single writer 写入调优测试。

1. Bloom Index。
2. upsert shuffle 并行度。
3. Parquet 文件大小和 small file 策略。
4. single writer 模式下的基础写入配置。
5. Timeline 和 HDFS 文件结构。

先准备测试脚本：

这里的 dt 是模拟业务分区日期，用于演示多分区写入和文件布局，不代表文章实际写作日期。ts 是模拟业务事件时间，用作 precombine 排序字段；dt 是模拟业务分区字段，本文只用于调优测试。

cat > /tmp/hudi_spark_tuning_demo.py <<'PY'
from pyspark.sql import SparkSession
from pyspark.sql.functions import count, max as spark_max, min as spark_min

spark = (
    SparkSession.builder
    .appName("hudi-spark-tuning-demo")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.hudi.catalog.HoodieCatalog")
    .getOrCreate()
)

base_path = "hdfs:///hudi/demo_spark_tuning_orders"
table_name = "demo_spark_tuning_orders"

rows = []
for i in range(2000):
    dt = "2025-08-27" if i % 2 == 0 else "2025-08-28"
    status = "PAID" if i % 3 == 0 else "CREATED"
    rows.append((900000 + i, "u%04d" % (i % 200), float(i % 500), status, 1756274700000 + i, dt))

df = spark.createDataFrame(
    rows,
    ["order_id", "user_id", "amount", "status", "ts", "dt"]
).repartition(4, "dt")

common_options = {
    "hoodie.table.name": table_name,
    "hoodie.datasource.write.recordkey.field": "order_id",
    "hoodie.datasource.write.partitionpath.field": "dt",
    "hoodie.datasource.write.precombine.field": "ts",
    "hoodie.datasource.write.table.type": "COPY_ON_WRITE",
    "hoodie.datasource.write.operation": "upsert",
    "hoodie.datasource.write.keygenerator.class": "org.apache.hudi.keygen.SimpleKeyGenerator",
    "hoodie.index.type": "BLOOM",
    "hoodie.metadata.enable": "true",
    "hoodie.upsert.shuffle.parallelism": "4",
    "hoodie.insert.shuffle.parallelism": "4",
    "hoodie.bulkinsert.shuffle.parallelism": "4",
    "hoodie.parquet.max.file.size": str(64 * 1024 * 1024),
    "hoodie.parquet.small.file.limit": str(32 * 1024 * 1024),
    "hoodie.copyonwrite.record.size.estimate": "512",
    "hoodie.write.concurrency.mode": "single_writer",
    "hoodie.write.lock.provider": "org.apache.hudi.client.transaction.lock.InProcessLockProvider"
}

print("=== tuning options ===")
for key in sorted(common_options):
    if key.startswith("hoodie.upsert") or key.startswith("hoodie.insert") \
            or key.startswith("hoodie.bulkinsert") or key.startswith("hoodie.parquet") \
            or key.startswith("hoodie.copyonwrite") or key.startswith("hoodie.index") \
            or key.startswith("hoodie.write"):
        print("%s=%s" % (key, common_options[key]))

print("=== input partitions ===")
print(df.rdd.getNumPartitions())

(
    df.write.format("hudi")
    .options(**common_options)
    .mode("overwrite")
    .save(base_path)
)

result = spark.read.format("hudi").load(base_path)
print("=== query result ===")
(
    result.groupBy("dt")
    .agg(
        count("*").alias("cnt"),
        spark_min("order_id").alias("min_order_id"),
        spark_max("order_id").alias("max_order_id")
    )
    .orderBy("dt")
    .show(truncate=False)
)

print("=== commit times ===")
result.select("_hoodie_commit_time").distinct().orderBy("_hoodie_commit_time").show(truncate=False)

spark.stop()
PY

本文 Hadoop 已配置默认文件系统，因此 hdfs:///hudi/demo_spark_tuning_orders 和 hdfs dfs -ls /hudi/demo_spark_tuning_orders 指向同一张 Hudi 表；如果没有配置 fs.defaultFS，需要写完整的 hdfs://host:port/...。本文使用 hoodie.table.name；如果你的版本要求 hoodie.datasource.write.table.name，请按当前 Hudi 配置文档调整。single_writer 是默认并发模式，这里显式写出只是为了和前面的 OCC 配置对比。如果没有启用多 writer / OCC，lock provider 可能不是当前测试的关键路径；本文只是展示单进程锁配置形态。

执行前清理旧目录：

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

这里同时删除旧目录并使用 mode("overwrite")，只是为了保证本地测试可重复；真实环境不要随意删除 Hudi 表目录。脚本里的 mode("overwrite") 是为了每次清理并重建测试表，hoodie.datasource.write.operation=upsert 表达 Hudi 写入语义。生产增量写入通常使用 mode("append")，不要把 Spark SaveMode 和 Hudi operation 混为一谈。

提交 Spark 任务：

spark-submit \
  --master local[2] \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension \
  --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog \
  --jars /opt/module/hudi-1.0.2/lib/hudi-spark3.5-bundle_2.12-1.0.2.jar \
  /tmp/hudi_spark_tuning_demo.py

核心输出如下：

=== tuning options ===
hoodie.bulkinsert.shuffle.parallelism=4
hoodie.copyonwrite.record.size.estimate=512
hoodie.index.type=BLOOM
hoodie.insert.shuffle.parallelism=4
hoodie.parquet.max.file.size=67108864
hoodie.parquet.small.file.limit=33554432
hoodie.upsert.shuffle.parallelism=4
hoodie.write.concurrency.mode=single_writer
hoodie.write.lock.provider=org.apache.hudi.client.transaction.lock.InProcessLockProvider
=== input partitions ===
4
=== query result ===
+----------+----+------------+------------+
|dt        |cnt |min_order_id|max_order_id|
+----------+----+------------+------------+
|2025-08-27|1000|900000      |901998      |
|2025-08-28|1000|900001      |901999      |
+----------+----+------------+------------+

=== commit times ===
+-------------------+
|_hoodie_commit_time|
+-------------------+
|20250828112712310  |
+-------------------+

这里有几个观察点：

1. 输入数据被 repartition 成 4 个分区。
2. 写入并行度也设置为 4，和本地测试规模匹配。
3. 只有一个 commit time，说明这次 overwrite 写入形成了一次完整提交。
4. 两个分区各 1000 条记录，符合预期。

2.3 查看 Timeline 和文件大小

查看 Timeline：

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

输出如下：

Found 4 items
-rw-r--r--   1 root supergroup          0 2025-08-28 11:27 /hudi/demo_spark_tuning_orders/.hoodie/timeline/20250828112712310.commit.requested
-rw-r--r--   1 root supergroup       3498 2025-08-28 11:27 /hudi/demo_spark_tuning_orders/.hoodie/timeline/20250828112712310.inflight
-rw-r--r--   1 root supergroup       4134 2025-08-28 11:27 /hudi/demo_spark_tuning_orders/.hoodie/timeline/20250828112712310_20250828112725263.commit
drwxr-xr-x   - root supergroup          0 2025-08-28 11:27 /hudi/demo_spark_tuning_orders/.hoodie/timeline/history

在 Hudi 1.0.x 中，完成态 Instant 文件名里可以看到 begin instant 和 completion instant，例如 begin_completion.commit；这里前半段更接近请求 / 开始时间，后半段是完成时间。这里是 Hudi 1.0.x Timeline 布局下的实测输出，requested / inflight 用于表达 Instant 状态流转；判断已经成功提交的动作时，主要看完成态 commit 文件。

查看数据文件：

hdfs dfs -ls -R /hudi/demo_spark_tuning_orders \
  | grep -E '2025-08-2[89]|parquet'

如果只想看 Parquet 文件，可以用更精确的过滤：

hdfs dfs -ls -R /hudi/demo_spark_tuning_orders | grep '.parquet$'

输出如下：

drwxr-xr-x   - root supergroup          0 2025-08-28 11:27 /hudi/demo_spark_tuning_orders/2025-08-27
-rw-r--r--   1 root supergroup     447979 2025-08-28 11:27 /hudi/demo_spark_tuning_orders/2025-08-27/76a4b954-c9da-4b5e-a733-554f18e439a6-0_0-37-55_20250828112712310.parquet
drwxr-xr-x   - root supergroup          0 2025-08-28 11:27 /hudi/demo_spark_tuning_orders/2025-08-28
-rw-r--r--   1 root supergroup     447994 2025-08-28 11:27 /hudi/demo_spark_tuning_orders/2025-08-28/d67f88e0-c1e5-4328-b21c-c742308f671d-0_1-37-56_20250828112712310.parquet

这次数据量很小，每个分区只形成一个 Parquet 文件。因为本文只有 2000 条小样本数据，每个分区数据量远小于 small file limit，所以文件不会达到 64MB 的目标大小；文件大小参数更多在持续写入或大批量数据中体现效果。生产环境里如果每个分区持续产生大量小文件，就需要结合 hoodie.parquet.small.file.limit、hoodie.parquet.max.file.size、bulk_insert 排序策略和 clustering 一起看。

再看表属性：

hdfs dfs -cat /hudi/demo_spark_tuning_orders/.hoodie/hoodie.properties \
  | grep -E 'hoodie.table.type|hoodie.table.version|hoodie.table.name|partition'

输出如下：

hoodie.table.type=COPY_ON_WRITE
hoodie.datasource.write.drop.partition.columns=false
hoodie.datasource.write.hive_style_partitioning=false
hoodie.datasource.write.partitionpath.urlencode=false
hoodie.table.version=8
hoodie.table.name=demo_spark_tuning_orders

本文没有开启 hoodie.datasource.write.hive_style_partitioning=true，因此分区目录直接是 2025-08-27、2025-08-28，而不是 dt=2025-08-27、dt=2025-08-28。

2.4 生产配置示例

下面参数不是推荐默认值，只是配置形态模板，必须根据数据规模、集群资源、分区数、目标文件大小和 Spark UI 结果调整。

如果是单任务写入，我通常从 single writer 开始：

hoodie.write.concurrency.mode=single_writer
# 单 writer 默认不需要分布式锁；如需同 JVM / 同进程协调，可使用 InProcessLockProvider
hoodie.upsert.shuffle.parallelism=200
hoodie.insert.shuffle.parallelism=200
hoodie.bulkinsert.shuffle.parallelism=200
hoodie.parquet.max.file.size=134217728
hoodie.parquet.small.file.limit=104857600
hoodie.copyonwrite.record.size.estimate=1024

如果需要多 writer，并且集群里已经有 ZooKeeper，可以使用类似配置：

hoodie.write.concurrency.mode=optimistic_concurrency_control
hoodie.cleaner.policy.failed.writes=LAZY
hoodie.write.lock.provider=org.apache.hudi.client.transaction.lock.ZookeeperBasedLockProvider
hoodie.write.lock.zookeeper.url=<zk-host>
hoodie.write.lock.zookeeper.port=2181
hoodie.write.lock.zookeeper.base_path=/hudi/locks

如果环境已经有 Hive Metastore，也可以考虑 HMS 锁：

hoodie.write.concurrency.mode=optimistic_concurrency_control
hoodie.cleaner.policy.failed.writes=LAZY
hoodie.write.lock.provider=org.apache.hudi.hive.transaction.lock.HiveMetastoreBasedLockProvider
hoodie.write.lock.hivemetastore.database=hudi
hoodie.write.lock.hivemetastore.table=orders

这里的 <zk-host>、hudi、orders 都只是示例，生产环境要替换成真实集群配置。HMS 锁配置需要以 Hudi 1.0.2 当前文档和实际 Hive Metastore 版本为准；本文前面测试环境中 Spark 3.5 与 Hive 4 HMS 存在兼容问题，不建议直接把这段作为已验证配置。多 writer 场景一定要做失败恢复测试，不要只验证“能写进去”。

2.4.1 Spark 提交参数模板

除了 Hudi 参数，生产 Spark 任务还要显式记录 Spark 资源和运行参数。下面这个模板更适合作为起点，而不是最终答案：

spark.serializer=org.apache.spark.serializer.KryoSerializer
spark.kryoserializer.buffer.max=512m
spark.driver.memory=4g
spark.driver.maxResultSize=2g
spark.executor.memory=6g
spark.executor.memoryOverhead=3g
spark.executor.cores=1
spark.executor.instances=<按集群资源和输入规模设置>
spark.task.cpus=1
spark.task.maxFailures=4
spark.rdd.compress=true
spark.shuffle.service.enabled=true
spark.sql.hive.convertMetastoreParquet=false
spark.executor.extraJavaOptions=-XX:+UseG1GC -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/tmp/hoodie-heapdump.hprof -Xlog:gc*:file=/tmp/spark-executor-gc-%p.log:time,uptime,level,tags:filecount=5,filesize=100M

这里有几个判断标准：

1. executor.cores=1 能降低单 executor 内多个 task 争抢内存的概率，但会增加 executor 数量，需要结合集群资源看。
2. executor.instances 需要按集群资源和输入规模设置，小测试环境不能照抄大集群参数。
3. spark.sql.hive.convertMetastoreParquet=false 主要影响 Spark 读取 Hive Metastore 中 Parquet 表时是否走 Spark 内置 Parquet 转换；是否和 Hudi 表读取 / 查询链路相关，要看你是通过 Hive Metastore 表、Hudi Catalog，还是 DataSource path 读取，不应默认所有 Hudi 写入任务都加。
4. extraJavaOptions 里的 heap dump 和 GC 日志路径要确保 executor 有写权限，并配置日志滚动和清理。

3. 常见问题与判断

3.1 常见问题

3.1.1 是否所有表都要开启 OCC

不需要。

如果一张表同一时刻只有一个 writer，single writer 更简单。OCC 适合多个任务可能同时写同一张表的场景。开启 OCC 以后，还要维护锁服务、失败写入清理策略和冲突重试逻辑。

3.1.2 加了锁是不是就不会冲突

不是。

锁主要保护提交阶段，让多个 writer 不要同时修改 Timeline 临界区。Hudi 还会做文件级冲突判断。如果两个 writer 都改了同一个 FileGroup，即使有锁，也可能有一个 writer 失败。

3.1.3 Upsert 写入很慢怎么办

先看四件事：

1. 输入数据量和 shuffle parallelism 是否匹配。
2. 索引是否太重，比如不必要地使用 Global Bloom。
3. 是否有大量小文件导致写入前要做很多 file sizing。
4. executor 是否频繁 spill 或 GC。

如果是初始化大表，不要一上来就用 upsert，可以优先考虑 bulk_insert。等基线数据导入完成后，再用 upsert 承接增量数据。

3.1.4 为什么会出现大量小文件

常见原因是每批数据太小、分区太细、写入并行度不合理，或者长期写入表没有配合 small file handling / clustering 做文件整理。MOR 表还需要额外关注 compaction，否则 log file 也会影响查询成本。

解决方向包括：

1. 合理设置 hoodie.parquet.max.file.size 和 hoodie.parquet.small.file.limit。
2. 批量初始化时使用 bulk_insert，并考虑按分区排序。
3. 对长期写入表启用 clustering。
4. 避免把分区字段切得过细。

3.1.5 OOM 该怎么排查

先判断 OOM 发生在 driver 还是 executor，然后看 Spark UI 里的 task spill、GC time、shuffle read/write、失败 stage。Hudi 写入里比较容易吃内存的地方包括索引定位、记录合并、Parquet 写入、MOR log merge、compaction。

排查方向：

spark.executor.memory=4g
spark.executor.memoryOverhead=1g
spark.memory.offHeap.enabled=true
spark.memory.offHeap.size=1g
hoodie.memory.merge.fraction=<按任务类型和版本文档调整>
hoodie.memory.compaction.fraction=<按任务类型和版本文档调整>

merge 和 compaction 通常不会在同一个 Spark action 中简单相加理解，但这类比例参数不能机械照抄，需要结合当前作业类型、Executor 内存和 Hudi 版本文档确认。是否启用 Spark off-heap 要单独压测；很多 Hudi 写入 OOM 更常见的根因是并行度、文件大小、索引定位、数据倾斜或单 task 数据过大，不要把 off-heap 当成万能解法。小集群调大内存可能有用，大集群里更重要的是数据分布、并行度和文件组织。

3.2 并发控制与调优判断

Hudi 的并发控制最容易被误解成“配置一个锁就行”。实际不是这样。锁只是提交阶段的一部分，真正让 Hudi 能在数据湖上稳定工作的，是 Timeline、MVCC、文件版本、失败写入清理和文件级冲突判断组合起来的结果。

Spark 调优也是类似。很多时候大家只盯着 hoodie.upsert.shuffle.parallelism，但这个参数只是调优入口。写入慢可能是索引定位慢，也可能是小文件太多，还可能是 executor 内存不足或者数据倾斜。Hudi 表越大，越要把“写入吞吐、查询延迟、文件数量、清理成本、索引成本”放在一起看。

我更倾向于把 Hudi 写入设计成两类链路：

1. 初始化链路：使用 bulk_insert，重点关注吞吐、排序和文件大小。
2. 增量链路：使用 upsert，重点关注索引、并发、checkpoint 和失败恢复。

如果把初始化和增量都混在一个长期运行任务里，后面调优会非常痛苦。

两条链路最好使用不同的参数模板和资源队列，不要用同一个长期任务同时承担初始化导入和日常 CDC 更新。

4. 总结

本文主要梳理了 Hudi 并发控制和 Spark 写入调优：

1. Hudi 通过 MVCC 让 reader 读取已完成快照，避免读写互相阻塞。
2. 多 writer 场景需要 OCC、锁服务和文件级冲突判断一起工作。
3. single writer 是最简单稳定的模型，不需要为了“高级”而强行开启 OCC。
4. Spark 写入调优要同时看 shuffle 并行度、索引成本、文件大小和内存。
5. 初始化大表优先考虑 bulk_insert，增量更新再使用 upsert。
6. DataFrame 和 HoodieStreamer 都可以加 OCC 参数，但前提是已经选好并验证了对应锁服务，例如 ZooKeeper、HMS 或云存储锁，并明确失败恢复和写入范围。
7. 小文件、OOM、写入慢通常不是孤立问题，要结合 Timeline、HDFS 文件和 Spark UI 一起分析。