Apache Hudi 学习笔记 08：使用 HoodieStreamer / DeltaStreamer 从 Kafka 导入数据到 Hudi

前面一篇已经用 Spark DataSource 和 Spark SQL 写过 Hudi 表。那种方式适合自己写 Spark 程序，但如果只是想快速把 Kafka、DFS、Sqoop 输出这类数据源导入 Hudi，Hudi 官方还提供了一个摄入工具：HoodieStreamer。

很多资料仍然把它叫 DeltaStreamer。这个叫法没有问题，因为历史上主类名确实是 HoodieDeltaStreamer。但在 Hudi 1.0.x 中，官方文档主线已经转向 org.apache.hudi.utilities.streamer.HoodieStreamer，旧的 org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer 类仍然存在，主要用于兼容旧任务。新文章和新脚本建议优先使用 HoodieStreamer。本文标题保留 DeltaStreamer，是为了和大家常用搜索关键词保持一致；正文实操统一使用新版 HoodieStreamer。

1. 摄入链路准备

1.1 背景问题

如果我们要把 Kafka 中的订单事件写入 Hudi，最直接的方式是自己写一个 Spark Streaming 或 Flink 程序：

Kafka -> Spark/Flink 任务 -> Hudi 表

这样做最灵活，但要自己处理不少工程细节：

1. 从 Kafka 哪个 offset 开始读。
2. 读取失败后如何恢复。
3. Schema 从哪里来。
4. 写入 Hudi 时用什么 record key、partition path、precombine field。
5. 是否要做 upsert、insert 或 bulk_insert。
6. 每次成功写入后，消费进度保存在哪里。

HoodieStreamer 的价值就在这里：它把这些通用摄入逻辑封装好了。我们只需要配置数据源、Schema、目标表和写入参数，就可以把 Kafka 数据导入 Hudi。

1.2 HoodieStreamer 能做什么

Hudi 1.0.2 文档中对 HoodieStreamer 的定位很明确：它是 Hudi utilities bundle 中的摄入工具，可以从 DFS、Kafka 等来源读取数据，支持 JSON、Avro、自定义格式，能够管理 checkpoint、rollback、recovery，也可以接入 schema provider 和 transformer。下面命令和配置以 Hudi 1.0.2 的 Hudi Streamer 文档与本文实测环境为准，latest 文档可能已有变化。

我把它理解成一个“官方通用入湖任务模板”：

能力	说明
Kafka Source	从 Kafka Topic 读取消息
DFS Source	从 HDFS/S3 路径读取文件
Schema Provider	从文件或 Schema Registry 获取 Avro Schema
Checkpoint	保存已消费 offset，下次继续读取
Write Operation	支持 UPSERT、INSERT、BULK_INSERT 等写入模式
Transformer	写入前可以做 SQL 或自定义转换
Hive Sync	可选同步 Hive Metastore

它不适合承载非常复杂的业务逻辑。如果要做多流 join、复杂维表关联、异步补偿、复杂 CDC 语义处理，我更倾向于写 Flink 或 Spark 作业。但如果目标是“快速把一个 Kafka Topic 可靠写入 Hudi 表”，HoodieStreamer 是非常合适的起点。

1.3 Kafka 到 Hudi 的整体链路

本文要跑的是下面这条链路：

这张图主要说明 Kafka 消息经过 JsonKafkaSource、FilebasedSchemaProvider 和 Hudi Spark Writer 后，最终以 commit 形式写入 Hudi 表。

核心流程是：

1. Kafka Producer 向 Topic 写入 JSON 订单事件。
2. HoodieStreamer 使用 JsonKafkaSource 从 Kafka 拉取 offset 范围。
3. FilebasedSchemaProvider 从本地文件读取 Avro Schema。
4. Hudi Spark Writer 按 UPSERT 写入 COW 表。
5. Hudi commit 中保存 checkpoint，下次从上次 offset 继续读。
6. Spark 查询 Hudi 表验证结果。

1.4 环境检查

先确认 Hudi utilities bundle 和 Spark bundle：

ls -lh \
  /opt/module/hudi-1.0.2/lib/hudi-spark3.5-bundle_2.12-1.0.2.jar \
  /opt/module/hudi-1.0.2/lib/hudi-utilities-bundle_2.12-1.0.2.jar

输出如下：

-rw-r--r-- 1 root root 106M Jun 15 21:36 /opt/module/hudi-1.0.2/lib/hudi-spark3.5-bundle_2.12-1.0.2.jar
-rw-r--r-- 1 root root 116M Jun 15 21:39 /opt/module/hudi-1.0.2/lib/hudi-utilities-bundle_2.12-1.0.2.jar

确认工具类存在：

jar tf /opt/module/hudi-1.0.2/lib/hudi-utilities-bundle_2.12-1.0.2.jar \
  | grep -E 'Hoodie(Streamer|DeltaStreamer)\.class|JsonKafkaSource|FilebasedSchemaProvider'

输出如下：

org/apache/hudi/utilities/config/FilebasedSchemaProviderConfig.class
org/apache/hudi/utilities/sources/JsonKafkaSource.class
org/apache/hudi/utilities/schema/FilebasedSchemaProvider.class
org/apache/hudi/utilities/sources/JsonKafkaSource$Config.class
org/apache/hudi/utilities/streamer/HoodieStreamer.class
org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.class

可以看到新旧主类都在。本文使用：

org.apache.hudi.utilities.streamer.HoodieStreamer

查看 --help：

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[1] \
  --conf spark.driver.extraClassPath=$CP \
  --conf spark.executor.extraClassPath=$CP \
  --class org.apache.hudi.utilities.streamer.HoodieStreamer \
  /opt/module/hudi-1.0.2/lib/hudi-utilities-bundle_2.12-1.0.2.jar \
  --help

关键参数节选如下：

--props
  path to properties file on localfs or dfs

--schemaprovider-class
  subclass of org.apache.hudi.utilities.schema.SchemaProvider

--source-class
  Built-in options:
  org.apache.hudi.utilities.sources.{JsonDFSSource, AvroDFSSource,
  JsonKafkaSource, AvroKafkaSource, HiveIncrPullSource}

--source-limit
  Maximum amount of data to read from source.
  DFS-Source => max bytes to read, Kafka-Source => max events to read

--op
  Takes one of these values: UPSERT, INSERT, BULK_INSERT,
  INSERT_OVERWRITE, INSERT_OVERWRITE_TABLE, DELETE_PARTITION

--target-base-path
  base path for the target hoodie table

--target-table
  name of the target table

完整参数以当前版本 spark-submit --help 输出为准。如果 --help 也遇到类加载问题，使用后面第 7 节的 CP 写法。

1.5 准备 Kafka Topic 和测试数据

创建独立 Topic：

/opt/module/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --create \
  --if-not-exists \
  --topic hudi_deltastreamer_orders \
  --partitions 1 \
  --replication-factor 1

如果 Topic 不存在，输出类似：

Created topic hudi_deltastreamer_orders.

查看 Topic：

/opt/module/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --describe \
  --topic hudi_deltastreamer_orders

输出如下：

Topic: hudi_deltastreamer_orders	TopicId: XTw9THOkSyeQ-TXoeClR7Q	PartitionCount: 1	ReplicationFactor: 1	Configs:
	Topic: hudi_deltastreamer_orders	Partition: 0	Leader: 1	Replicas: 1	Isr: 1

准备第一批 JSON 消息：

cat > /tmp/hudi_orders_kafka.jsonl <<'JSONL'
{"order_id":8001,"user_id":"u501","amount":99.8,"status":"CREATED","ts":1755266820000,"dt":"2025-08-15"}
{"order_id":8002,"user_id":"u502","amount":168.0,"status":"CREATED","ts":1755266880000,"dt":"2025-08-15"}
{"order_id":8003,"user_id":"u503","amount":39.9,"status":"CREATED","ts":1755266940000,"dt":"2025-08-16"}
{"order_id":8002,"user_id":"u502","amount":168.0,"status":"PAID","ts":1755267000000,"dt":"2025-08-15"}
JSONL

这里的 dt 是模拟业务分区日期，用于演示跨分区写入，不代表文章实际写作日期；ts 是模拟业务事件时间，用作 precombine 排序字段。

写入 Kafka：

/opt/module/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic hudi_deltastreamer_orders \
  < /tmp/hudi_orders_kafka.jsonl

消费验证：

/opt/module/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic hudi_deltastreamer_orders \
  --from-beginning \
  --max-messages 4 \
  --timeout-ms 10000

消费输出类似如下，最后会看到 Processed a total of 4 messages：

{"order_id":8001,"user_id":"u501","amount":99.8,"status":"CREATED","ts":1755266820000,"dt":"2025-08-15"}
{"order_id":8002,"user_id":"u502","amount":168.0,"status":"CREATED","ts":1755266880000,"dt":"2025-08-15"}
{"order_id":8003,"user_id":"u503","amount":39.9,"status":"CREATED","ts":1755266940000,"dt":"2025-08-16"}
{"order_id":8002,"user_id":"u502","amount":168.0,"status":"PAID","ts":1755267000000,"dt":"2025-08-15"}
Processed a total of 4 messages

这里故意给 8002 写了两条消息，后一条 ts 更大，状态是 PAID。后面验证 Hudi 表时，应该只保留 8002 的最新状态。

1.5.1 如果要用 Java Producer 造数

这篇为了可复制，直接使用 kafka-console-producer.sh 写 JSONL。真实压测或批量造数时，可以写一个简单 Java Producer。

Maven 依赖示例：

<dependencies>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>3.9.1</version>
    </dependency>
</dependencies>

Producer 参数可以这样写：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("batch.size", "1048576");
props.put("linger.ms", "5");
props.put("compression.type", "snappy");
props.put("buffer.memory", "33554432");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

这里几个参数的含义：

参数	说明
acks=all	等待 ISR 中副本确认，适合更稳妥的写入
batch.size	单个分区 producer batch 大小
linger.ms	等待更多消息合并成批次，提升吞吐
compression.type=snappy	降低网络传输量和 Kafka 磁盘占用
buffer.memory	Producer 客户端缓冲区大小

如果只是验证 HoodieStreamer 链路，控制台生产者更直接；如果要压测 source-limit、commit 间隔、小文件和写入吞吐，Java Producer 更合适。

1.6 准备 Schema 和配置文件

HoodieStreamer 的配置关系如下：

这张图主要说明 HoodieStreamer 命令本身只负责指定入口类和参数，真正的数据源、schema、record key、partition path、precombine 等配置主要来自 props 文件。

创建配置目录：

mkdir -p /opt/module/hudi-1.0.2/streamer-conf

准备 source schema：

cat > /opt/module/hudi-1.0.2/streamer-conf/order_source.avsc <<'JSON'
{
  "type": "record",
  "name": "order_event",
  "namespace": "cn.hnbian.hudi",
  "fields": [
    {"name": "order_id", "type": "long"},
    {"name": "user_id", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "status", "type": "string"},
    {"name": "ts", "type": "long"},
    {"name": "dt", "type": "string"}
  ]
}
JSON

这里 source schema 和 target schema 保持一致：

cp \
  /opt/module/hudi-1.0.2/streamer-conf/order_source.avsc \
  /opt/module/hudi-1.0.2/streamer-conf/order_target.avsc

准备 Kafka Source 配置：

cat > /opt/module/hudi-1.0.2/streamer-conf/kafka-source.properties <<'PROPS'
bootstrap.servers=localhost:9092
group.id=hudi-streamer-orders-g1
auto.offset.reset=earliest

hoodie.streamer.source.kafka.topic=hudi_deltastreamer_orders
hoodie.streamer.schemaprovider.source.schema.file=file:///opt/module/hudi-1.0.2/streamer-conf/order_source.avsc
hoodie.streamer.schemaprovider.target.schema.file=file:///opt/module/hudi-1.0.2/streamer-conf/order_target.avsc

hoodie.datasource.write.recordkey.field=order_id
hoodie.datasource.write.partitionpath.field=dt
hoodie.datasource.write.precombine.field=ts
hoodie.datasource.write.keygenerator.class=org.apache.hudi.keygen.SimpleKeyGenerator
hoodie.datasource.write.table.type=COPY_ON_WRITE
hoodie.metadata.enable=true
PROPS

配置文件里最关键的是这几类：

配置	作用
bootstrap.servers	Kafka 地址
group.id	Kafka 消费组
auto.offset.reset	没有 checkpoint 时从哪里开始读
hoodie.streamer.source.kafka.topic	Streamer 读取的 Topic
hoodie.streamer.schemaprovider.*.schema.file	source 和 target Avro Schema
hoodie.datasource.write.recordkey.field	Hudi record key / 业务主键字段
hoodie.datasource.write.partitionpath.field	Hudi 分区字段
hoodie.datasource.write.precombine.field	同 key 取最新记录的比较字段

本文使用 Hudi 1.0.2 的 hoodie.streamer.* 配置；如果使用旧版 HoodieDeltaStreamer 或旧版本 Hudi，需要核对是否仍使用 hoodie.deltastreamer.* 配置 key。JsonKafkaSource 会按 Hudi source 实现读取 Kafka 消息值中的 JSON；本文没有额外配置 Kafka deserializer，是因为当前 Hudi 1.0.2 JsonKafkaSource 实测不需要手动指定。若换成 AvroKafkaSource、Schema Registry 或自定义 Source，需要按对应 source 补充配置。

1.6.1 配置文件拆分：base.properties 和 source.properties

本文为了让命令集中可读，把 Kafka、Schema 和 Hudi 写入参数放在一个 kafka-source.properties 中。下面的拆分方式是生产建议 / 扩展示例，不作为本文主线实测命令。生产里我更建议拆成两层：

base.properties
kafka-source.properties

base.properties 放所有 Topic 通用的 Hudi 写入参数：

cat > /opt/module/hudi-1.0.2/streamer-conf/base.properties <<'PROPS'
hoodie.datasource.write.keygenerator.class=org.apache.hudi.keygen.SimpleKeyGenerator
hoodie.datasource.write.table.type=COPY_ON_WRITE
hoodie.datasource.write.precombine.field=ts
hoodie.metadata.enable=true
hoodie.clean.automatic=true
hoodie.clean.async=true
PROPS

kafka-source.properties 只放当前 Topic 自己的 source、schema 和 key 配置：

cat > /opt/module/hudi-1.0.2/streamer-conf/kafka-source-split.properties <<'PROPS'
include=file:///opt/module/hudi-1.0.2/streamer-conf/base.properties

bootstrap.servers=localhost:9092
group.id=hudi-streamer-orders-g1
auto.offset.reset=earliest

hoodie.streamer.source.kafka.topic=hudi_deltastreamer_orders
hoodie.streamer.schemaprovider.source.schema.file=file:///opt/module/hudi-1.0.2/streamer-conf/order_source.avsc
hoodie.streamer.schemaprovider.target.schema.file=file:///opt/module/hudi-1.0.2/streamer-conf/order_target.avsc

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

如果配置和 schema 放到 HDFS，include 和 schema 路径也可以使用 HDFS 地址：

include=hdfs://localhost:9000/hudi-props/base.properties
hoodie.streamer.schemaprovider.source.schema.file=hdfs://localhost:9000/hudi-props/order_source.avsc
hoodie.streamer.schemaprovider.target.schema.file=hdfs://localhost:9000/hudi-props/order_target.avsc

上传命令示例：

hdfs dfs -mkdir -p /hudi-props
hdfs dfs -put -f /opt/module/hudi-1.0.2/streamer-conf/base.properties /hudi-props/
hdfs dfs -put -f /opt/module/hudi-1.0.2/streamer-conf/order_source.avsc /hudi-props/
hdfs dfs -put -f /opt/module/hudi-1.0.2/streamer-conf/order_target.avsc /hudi-props/
hdfs dfs -put -f /opt/module/hudi-1.0.2/streamer-conf/kafka-source-split.properties /hudi-props/

本地文件适合单机测试，HDFS 配置适合 YARN 提交，因为 driver 和 executor 都能从统一路径读取配置。如果使用 include=file:///... 或 HDFS include，需要先在当前 Hudi 版本上单独验证配置加载行为；不确定时，可以用外部脚本或配置管理工具先合并 base.properties 与 source properties，再交给 --props。

确认文件：

ls -l /opt/module/hudi-1.0.2/streamer-conf

输出如下：

total 20
-rw-r--r-- 1 root root 186 Aug 16 22:08 base.properties
-rw-r--r-- 1 root root 680 Aug 16 22:08 kafka-source.properties
-rw-r--r-- 1 root root 548 Aug 16 22:08 kafka-source-split.properties
-rw-r--r-- 1 root root 342 Aug 16 22:08 order_source.avsc
-rw-r--r-- 1 root root 342 Aug 16 22:08 order_target.avsc

2. HoodieStreamer 实测

2.1 第一次运行 HoodieStreamer

目标表路径：

/hudi/demo_deltastreamer_orders

本文 Spark/Hadoop 已配置默认文件系统，因此 /hudi/demo_deltastreamer_orders 指向 HDFS 路径；如果没有配置 fs.defaultFS，需要写成 hdfs://localhost:9000/hudi/demo_deltastreamer_orders。

第一次运行前先清理旧表：

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

为了保证本文结果可复现，最好使用全新的 Kafka Topic，或者先删除并重建 Topic。仅删除 Hudi 表只会清理 Hudi checkpoint，不会清理 Kafka Topic 中已有消息。

运行命令如下：

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.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog \
  --conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension \
  --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 4

本文使用 driver / executor extraClassPath 同时放入 utilities bundle 和 Spark bundle，是因为 HoodieStreamer 运行时会加载 utilities 内部类和 Hudi Spark 写入类；如果只把 utilities bundle 当主 jar，可能遇到类加载问题。后续命令继续复用当前 shell 中的 CP 变量；如果新开终端，需要重新执行上面的 CP=...。

几个参数解释：

参数	说明
--props	指定 Kafka、Schema、Hudi 写入参数
--schemaprovider-class	使用文件方式加载 Avro Schema
--source-class	使用 JSON Kafka Source
--source-ordering-field	同 key 多条记录按 ts 取最新
--target-base-path	Hudi 表 HDFS 路径
--target-table	Hudi 表名
--table-type	COW 表
--op	使用 UPSERT
--source-limit	本轮最多读取 4 条 Kafka 消息

source-limit 限制的是本轮最多读取的事件数，不是起始或结束 offset；实际 offset 范围由已有 checkpoint、Kafka Topic offset 和 source-limit 共同决定。

关键日志如下：

25/08/16 22:09:27 INFO StreamSync: Checkpoint to resume from : Optional.empty
25/08/16 22:09:28 INFO KafkaOffsetGen: final ranges [OffsetRange(topic: 'hudi_deltastreamer_orders', partition: 0, range: [0 -> 4])]
25/08/16 22:09:28 INFO KafkaSource: About to read sourceLimit 4 in 0 spark partitions from kafka for topic hudi_deltastreamer_orders with offset ranges [OffsetRange(topic: 'hudi_deltastreamer_orders', partition: 0, range: [0 -> 4])]
25/08/16 22:09:28 INFO KafkaSource: About to read 4 from Kafka for topic :hudi_deltastreamer_orders after offset generation with offset ranges [OffsetRange(topic: 'hudi_deltastreamer_orders', partition: 0, range: [0 -> 4])]
25/08/16 22:09:38 INFO StreamSync: Commit 20250816220927673 successful!

在本文这个新表、新 Topic、auto.offset.reset=earliest 的测试场景中，第一次运行没有历史 checkpoint，因此从 Kafka offset 0 开始读，最终读取 [0 -> 4] 四条消息，并提交一个 Hudi commit。

2.1.1 BULK_INSERT + MOR 适合什么场景

旧版资料中经常用 --op BULK_INSERT --table-type MERGE_ON_READ 做演示，这个组合不是不能用，只是语义和本文主线不同。

BULK_INSERT 更适合已经清洗好、record key 唯一性有保障的初始化导入。如果同一批数据中存在重复 key 或需要按 precombine 取最新，应该先在上游去重，或使用 upsert 语义验证结果。组合形态类似：

spark-submit \
  --master local[2] \
  --conf spark.driver.extraClassPath=$CP \
  --conf spark.executor.extraClassPath=$CP \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --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_mor \
  --target-table demo_deltastreamer_orders_mor \
  --table-type MERGE_ON_READ \
  --op BULK_INSERT \
  --source-limit 1000

这个命令只是展示组合形态。如果要实际跑，建议复制一份 properties，使用新的 group.id、新的 target-base-path，并确认是否需要使用当前版本支持的 checkpoint 重放参数或清理旧表，避免和主线任务互相影响。

我在这篇主线里选择 UPSERT + COPY_ON_WRITE，是因为 Kafka 测试数据里故意包含了重复 order_id，需要验证 precombine.field=ts 的更新效果。如果上游只是纯追加日志，BULK_INSERT 的写入成本会更低；如果后续还会持续更新，还是要回到 UPSERT 或基于 CDC 语义设计写入方式。

2.2 查询第一次入湖结果

查询脚本：

这里的 /hudi/... 是 HDFS 默认文件系统路径，不是本地目录。

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

spark = (SparkSession.builder
    .appName("hudi-streamer-verify")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.hudi.catalog.HoodieCatalog")
    .config("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
    .config("spark.kryo.registrator", "org.apache.spark.HoodieSparkKryoRegistrar")
    .getOrCreate())

spark.sparkContext.setLogLevel("ERROR")

base = "/hudi/demo_deltastreamer_orders"
df = spark.read.format("hudi").load(base)
df.createOrReplaceTempView("orders")

print("=== hudi table result ===")
spark.sql("""
SELECT order_id, user_id, amount, status, dt, _hoodie_commit_time
FROM orders
ORDER BY order_id
""").show(truncate=False)

print("=== count ===")
spark.sql("SELECT count(*) AS cnt FROM orders").show(truncate=False)

spark.stop()
PY

执行：

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

第一次查询结果如下：

=== hudi table result ===
+--------+-------+------+-------+----------+-------------------+
|order_id|user_id|amount|status |dt        |_hoodie_commit_time|
+--------+-------+------+-------+----------+-------------------+
|8001    |u501   |99.8  |CREATED|2025-08-15|20250816220927673  |
|8002    |u502   |168.0 |PAID   |2025-08-15|20250816220927673  |
|8003    |u503   |39.9  |CREATED|2025-08-16|20250816220927673  |
+--------+-------+------+-------+----------+-------------------+

=== count ===
+---+
|cnt|
+---+
|3  |
+---+

Kafka 里有 4 条消息，但 Hudi 表里只有 3 条业务记录，是因为 8002 出现了两次，后一条 ts 更大，状态为 PAID。这说明 precombine.field=ts 和 UPSERT 生效了。

2.2.1 使用 Spark SQL 注册已有 Hudi 表

除了 DataFrame 读取，也可以在 Spark SQL 中把已有 Hudi 路径注册成表再查询：

CREATE DATABASE IF NOT EXISTS hudi_demo;
USE hudi_demo;

DROP TABLE IF EXISTS demo_deltastreamer_orders;

CREATE TABLE demo_deltastreamer_orders
USING HUDI
LOCATION '/hudi/demo_deltastreamer_orders';

SELECT order_id, user_id, amount, status, dt, _hoodie_commit_time
FROM demo_deltastreamer_orders
ORDER BY order_id;

执行方式：

cat > /tmp/hudi_streamer_query.sql <<'SQL'
CREATE DATABASE IF NOT EXISTS hudi_demo;
USE hudi_demo;

DROP TABLE IF EXISTS demo_deltastreamer_orders;

CREATE TABLE demo_deltastreamer_orders
USING HUDI
LOCATION '/hudi/demo_deltastreamer_orders';

SELECT order_id, user_id, amount, status, dt, _hoodie_commit_time
FROM demo_deltastreamer_orders
ORDER BY order_id;
SQL

spark-sql \
  --master local[2] \
  --jars /opt/module/hudi-1.0.2/lib/hudi-spark3.5-bundle_2.12-1.0.2.jar \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --conf spark.sql.catalogImplementation=in-memory \
  --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog \
  --conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension \
  --conf spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar \
  -f /tmp/hudi_streamer_query.sql

这里使用 catalogImplementation=in-memory 是为了避开 Spark 3.5 和 Hive 4 Metastore 的兼容性问题，这只是本文单机测试策略。生产环境需要单独验证 Spark、Hive Metastore、Hudi Catalog / Hive Sync 兼容组合，后面第 15 篇会单独讲 Hive Sync 和 Hive Metastore 管理方式。

这里使用 in-memory catalog 注册已有路径，DROP TABLE 主要用于清理当前 Spark SQL 会话中的表定义，不代表删除 HDFS 上的 Hudi 表数据。真实 Metastore 环境下 DROP TABLE 的行为要按表类型和 Catalog 配置确认。如果已有路径是完整 Hudi 表，Hudi 通常可以从 .hoodie/hoodie.properties 读取表配置；如果当前版本或 Catalog 无法自动识别，需要补充 primaryKey、preCombineField、type 等 TBLPROPERTIES，或通过 Hive Sync / Metastore 统一注册。

2.3 第二次运行：验证 checkpoint

HoodieStreamer checkpoint 的续跑逻辑如下：

这张图主要说明 checkpoint 保存在 Hudi commit 元数据中，下一次运行会从上一次提交的 Kafka offset 后继续读取。

再写入第二批 Kafka 消息：

cat > /tmp/hudi_orders_kafka_2.jsonl <<'JSONL'
{"order_id":8004,"user_id":"u504","amount":258.6,"status":"CREATED","ts":1755267240000,"dt":"2025-08-16"}
{"order_id":8001,"user_id":"u501","amount":99.8,"status":"PAID","ts":1755267300000,"dt":"2025-08-15"}
JSONL

/opt/module/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic hudi_deltastreamer_orders \
  < /tmp/hudi_orders_kafka_2.jsonl

第二批中 8001 的 ts 更大，所以 upsert 后会覆盖第一次的 CREATED 状态。

第二次运行 HoodieStreamer。命令和第一次基本一致，只是 --source-limit 放宽到 10：

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.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog \
  --conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension \
  --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 10

关键日志：

25/08/16 22:10:43 INFO StreamSync: Checkpoint to resume from : Option{val=StreamerCheckpointV2{checkpointKey='hudi_deltastreamer_orders,0:4'}}
25/08/16 22:10:43 INFO KafkaOffsetGen: final ranges [OffsetRange(topic: 'hudi_deltastreamer_orders', partition: 0, range: [4 -> 6])]
25/08/16 22:10:43 INFO KafkaSource: About to read sourceLimit 10 in 0 spark partitions from kafka for topic hudi_deltastreamer_orders with offset ranges [OffsetRange(topic: 'hudi_deltastreamer_orders', partition: 0, range: [4 -> 6])]
25/08/16 22:10:43 INFO KafkaSource: About to read 2 from Kafka for topic :hudi_deltastreamer_orders after offset generation with offset ranges [OffsetRange(topic: 'hudi_deltastreamer_orders', partition: 0, range: [4 -> 6])]
25/08/16 22:10:53 INFO StreamSync: Commit 20250816221043454 successful!

这段日志很关键：

1. 第一次 commit 保存了 checkpoint 0:4。
2. 第二次运行从 0:4 继续。
3. Kafka 本次读取 [4 -> 6]，也就是两条新消息。
4. 第二次写入成功提交 20250816221043454。

这里的 checkpoint 是 HoodieStreamer 写入 Hudi commit 元数据中的 source checkpoint，不要简单等同于 Kafka consumer group 在 Kafka broker 中提交的 offset。排查时应同时看 Hudi commit metadata 和 Kafka group 状态。

最终查询结果：

=== hudi table result ===
+--------+-------+------+-------+----------+-------------------+
|order_id|user_id|amount|status |dt        |_hoodie_commit_time|
+--------+-------+------+-------+----------+-------------------+
|8001    |u501   |99.8  |PAID   |2025-08-15|20250816221043454  |
|8002    |u502   |168.0 |PAID   |2025-08-15|20250816220927673  |
|8003    |u503   |39.9  |CREATED|2025-08-16|20250816220927673  |
|8004    |u504   |258.6 |CREATED|2025-08-16|20250816221043454  |
+--------+-------+------+-------+----------+-------------------+

=== count ===
+---+
|cnt|
+---+
|4  |
+---+

可以看到：

1. 8001 被第二批消息更新为 PAID。
2. 8004 被新增。
3. 8002 仍然是第一次写入后的 PAID。
4. 表里最终有 4 条业务记录。

2.4 查看 Hudi Timeline 和文件

查看 Timeline：

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

输出如下：

Found 7 items
-rw-r--r--   1 root supergroup          0 2025-08-16 22:09 /hudi/demo_deltastreamer_orders/.hoodie/timeline/20250816220927673.commit.requested
-rw-r--r--   1 root supergroup       3496 2025-08-16 22:09 /hudi/demo_deltastreamer_orders/.hoodie/timeline/20250816220927673.inflight
-rw-r--r--   1 root supergroup       4126 2025-08-16 22:09 /hudi/demo_deltastreamer_orders/.hoodie/timeline/20250816220927673_20250816220938453.commit
-rw-r--r--   1 root supergroup          0 2025-08-16 22:10 /hudi/demo_deltastreamer_orders/.hoodie/timeline/20250816221043454.commit.requested
-rw-r--r--   1 root supergroup       3600 2025-08-16 22:10 /hudi/demo_deltastreamer_orders/.hoodie/timeline/20250816221043454.inflight
-rw-r--r--   1 root supergroup       4312 2025-08-16 22:10 /hudi/demo_deltastreamer_orders/.hoodie/timeline/20250816221043454_20250816221053426.commit
drwxr-xr-x   - root supergroup          0 2025-08-16 22:09 /hudi/demo_deltastreamer_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_deltastreamer_orders | sed -n '1,90p'

关键输出如下：

drwxr-xr-x   - root supergroup          0 2025-08-16 22:09 /hudi/demo_deltastreamer_orders/.hoodie
-rw-r--r--   1 root supergroup       1193 2025-08-16 22:09 /hudi/demo_deltastreamer_orders/.hoodie/hoodie.properties
drwxr-xr-x   - root supergroup          0 2025-08-16 22:09 /hudi/demo_deltastreamer_orders/.hoodie/timeline
-rw-r--r--   1 root supergroup       4126 2025-08-16 22:09 /hudi/demo_deltastreamer_orders/.hoodie/timeline/20250816220927673_20250816220938453.commit
-rw-r--r--   1 root supergroup       4312 2025-08-16 22:10 /hudi/demo_deltastreamer_orders/.hoodie/timeline/20250816221043454_20250816221053426.commit
drwxr-xr-x   - root supergroup          0 2025-08-16 22:09 /hudi/demo_deltastreamer_orders/2025-08-15
-rw-r--r--   1 root supergroup     435549 2025-08-16 22:09 /hudi/demo_deltastreamer_orders/2025-08-15/76128298-ffed-4fc1-a5ea-00d55624b0a6-0_0-31-26_20250816220927673.parquet
drwxr-xr-x   - root supergroup          0 2025-08-16 22:09 /hudi/demo_deltastreamer_orders/2025-08-16
-rw-r--r--   1 root supergroup     435405 2025-08-16 22:09 /hudi/demo_deltastreamer_orders/2025-08-16/ac99df60-c114-486f-ae55-a4728fa9be85-0_1-31-27_20250816220927673.parquet

上面截取主要展示了表目录和第一次 commit 的 Base File。第二次运行更新 8001 并新增 8004 后，完整输出中还能看到第二次 commit 对应的新 Base File，可以用下面命令单独过滤：

hdfs dfs -ls -R /hudi/demo_deltastreamer_orders | grep 20250816221043454

输出类似：

-rw-r--r--   1 root supergroup     435738 2025-08-16 22:10 /hudi/demo_deltastreamer_orders/2025-08-15/76128298-ffed-4fc1-a5ea-00d55624b0a6-0_0-45-39_20250816221043454.parquet
-rw-r--r--   1 root supergroup     435487 2025-08-16 22:10 /hudi/demo_deltastreamer_orders/2025-08-16/ac99df60-c114-486f-ae55-a4728fa9be85-0_1-45-40_20250816221043454.parquet

从结果可以看出：

1. 目标表是标准 Hudi 表，有 .hoodie 和 hoodie.properties。
2. Timeline 中有两次 commit，对应两次 Streamer 运行。
3. COW 表的数据落在分区目录下，文件是 Parquet Base File。
4. 完整文件列表中可以看到两次 commit 对应的 Base File，Snapshot Query 会按 Timeline 选择当前有效的 FileSlice。

3. 常见问题与使用边界

3.1 常见问题

3.1.1 HoodieStreamer 和 HoodieDeltaStreamer 到底用哪个

Hudi 1.0.x 推荐使用：

org.apache.hudi.utilities.streamer.HoodieStreamer

旧类名：

org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer

仍然存在，但新文章和新脚本建议优先使用 HoodieStreamer。如果线上旧任务还在使用 HoodieDeltaStreamer，升级前要先确认命令参数和配置项是否需要调整。

3.1.2 ClassNotFoundException: HoodieIngestionService$HoodieIngestionConfig

这次我第一次运行时遇到了真实报错：

Exception in thread "main" org.apache.hudi.exception.HoodieException: Failed to run HoodieStreamer
Caused by: java.lang.ClassNotFoundException:
org.apache.hudi.utilities.ingestion.HoodieIngestionService$HoodieIngestionConfig

但检查 jar 时，这个类是存在的：

org/apache/hudi/utilities/ingestion/HoodieIngestionService$HoodieIngestionConfig.class

在本文环境中，jar 内确实存在该类，因此这次问题不是 utilities bundle 本身缺类，而更像是 Spark 启动时主 jar 与运行时 classloader 加载位置不一致，或 Hudi bundle 加载顺序不完整。其他环境还要同时排查是否混用了多个 Hudi 版本。最终修复方式是把 utilities bundle 和 Spark bundle 同时加入 driver / executor classpath：

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

--conf spark.driver.extraClassPath=$CP
--conf spark.executor.extraClassPath=$CP

修复后同样命令可以正常提交并生成 commit。

同时确认 Spark jars、--jars、extraClassPath 中不要混入多个不同版本的 hudi-spark-bundle、hudi-utilities-bundle，否则可能出现更隐蔽的类冲突。

3.1.3 Kafka 有数据但 HoodieStreamer 没读到

优先检查：

1. hoodie.streamer.source.kafka.topic 是否写对。
2. bootstrap.servers 是否能连通。
3. auto.offset.reset 是否符合预期。
4. Hudi 表里是否已有 checkpoint。
5. 是否需要使用当前版本 HoodieStreamer 支持的 checkpoint 相关参数进行重放，例如 --ignore-checkpoint 或 --checkpoint。具体参数名以本文第 4 节 --help 输出为准。

如果日志中出现：

Checkpoint to resume from : Option{val=StreamerCheckpointV2{checkpointKey='hudi_deltastreamer_orders,0:4'}}

说明它不会从头读，而是从已有 checkpoint 后继续。

3.1.4 JSON 字段和 Avro Schema 不一致

如果 Kafka JSON 缺字段、字段类型不匹配，或者 Avro Schema 写错，Streamer 可能在解析阶段失败，或者写入时出现 schema 兼容问题。

排查顺序：

1. 先用 kafka-console-consumer 看原始 JSON。
2. 确认 JSON 字段名和 Avro Schema 字段名一致。
3. 确认 recordkey.field、partitionpath.field、precombine.field 都在 Schema 中存在。
4. 如果上游字段可能缺失，要考虑默认值、nullable schema，或者先用 transformer 清洗。

nullable 字段可以写成类似下面这种 Avro 形式：

{"name":"status","type":["null","string"],"default":null}

3.1.5 source-limit 是什么

--source-limit 对 Kafka Source 表示本次最多读取多少条事件。学习和调试时建议设置小一点，便于观察 offset 范围和 commit。本文只演示一次性 batch 运行；continuous 模式或定时调度运行会在后续生产化文章中单独验证。

3.1.6 hoodie.deltastreamer.* 和 hoodie.streamer.* 怎么选

旧版本示例里经常能看到：

hoodie.deltastreamer.source.kafka.topic=hudi_test
hoodie.deltastreamer.schemaprovider.source.schema.file=...

Hudi 1.0.x 新示例更推荐使用 hoodie.streamer.*：

hoodie.streamer.source.kafka.topic=hudi_deltastreamer_orders
hoodie.streamer.schemaprovider.source.schema.file=file:///opt/module/hudi-1.0.2/streamer-conf/order_source.avsc
hoodie.streamer.schemaprovider.target.schema.file=file:///opt/module/hudi-1.0.2/streamer-conf/order_target.avsc

如果从旧任务升级，不要只替换主类名，还要一起检查配置 key。同一个任务中不要同时混用 hoodie.deltastreamer.* 和 hoodie.streamer.* 的同义配置，避免最终生效值不清楚。最稳妥的方式是在测试环境先跑 --help，再用一小批 Kafka 消息验证读取 offset、schema 解析和 commit 元数据。

3.2 HoodieStreamer 使用边界

HoodieStreamer 的核心价值不是“少写几行代码”，而是把摄入任务中最容易遗漏的通用机制做成了标准流程：source、schema、checkpoint、write config、commit、rollback。

自己写 Spark 任务当然更灵活，但灵活也意味着要自己维护更多状态。比如 Kafka offset 保存在哪里、写入失败后是否重复消费、Schema 怎么演进、同 key 多条记录怎么取最新，这些都不是几行 readStream 和 write 能自然解决的问题。

我更愿意把 HoodieStreamer 当作快速入湖工具：新 Topic 接入、简单 JSON/Avro 事件入湖、原型验证、数据链路打通，都可以优先用它。等业务逻辑复杂到需要多流处理、强 CDC 语义或者复杂转换时，再切到 Flink/Spark 自定义任务会更合适。这里的强 CDC 语义包括 before / after、op 类型、delete 事件、乱序事件和幂等 merge 等，不只是把 source 换成 Kafka。

另外，这篇实测里暴露出的 classpath 问题也说明：Hudi utilities 工具虽然封装了业务逻辑，但仍然运行在 Spark 生态里。Spark、Hudi bundle、utilities bundle、Kafka client、Hadoop classpath 这些版本和加载顺序，仍然要认真对待。

4. 总结

这一篇完成了 Kafka 到 Hudi 的 HoodieStreamer 实战：

1. Hudi 1.0.x 中建议使用 HoodieStreamer，DeltaStreamer 是更常见的旧称。
2. Kafka JSON 数据可以通过 JsonKafkaSource 导入 Hudi。
3. FilebasedSchemaProvider 需要 source schema 和 target schema。
4. recordkey.field、partitionpath.field、precombine.field 是写入正确性的关键。
5. 第一次运行从 Kafka offset [0 -> 4] 读取 4 条消息，写入后得到 3 条业务记录。
6. 第二次运行从 checkpoint 0:4 继续读取 [4 -> 6] 两条新消息。
7. Hudi Timeline 中能看到两次 commit，对应两次 Streamer 运行。
8. base.properties、HDFS props、Spark SQL location 注册表，是从单机测试走向统一配置、统一查询时经常会遇到的补充能力；生产环境更推荐结合 Hive Sync / Metastore / Catalog 方案统一管理表。
9. 如果遇到 utilities bundle 类加载问题，需要检查 driver/executor classpath。