环境准备

Spark 作业使用 Hudi 前需引入 Hudi Spark Bundle（含 DataSource 与 SQL 扩展），并启用 Hudi 的 SQL 扩展与 Catalog。以 spark-submit 提交作业为例，配置方式如下：

spark-submit \
  --master yarn --deploy-mode client \
  --jars /usr/local/spark/jars/hudi-spark-bundle-1.2.0-1.0.0.jar \
  --conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension \
  --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog \
  --conf spark.sql.catalog.spark_catalog.impl=org.apache.spark.sql.hudi.catalog.HoodieCatalog \
  <application-jar>

交互式场景使用spark-sql 客户端时，启动命令携带相同参数：

spark-sql \
  --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' \
  --conf 'spark.hadoop.spark.sql.legacy.parquet.nanosAsLong=false'

各参数含义说明如下表所示：

  --conf spark.speculation=false \
  --conf spark.sql.defaultColumn.enabled=false \

两个参数的强制原因如下表所示（依据Hudi 1.2.0 源码）：

创建表

使用Spark SQL 创建 Hudi 表，通过 TBLPROPERTIES 指定表类型、主键与预合并字段：

CREATE TABLE hudi_db.orders (
  order_id STRING,
  order_date STRING,
  amount DOUBLE,
  ts BIGINT
)
USING hudi
PARTITIONED BY (order_date)
TBLPROPERTIES (
  type = 'cow',
  primaryKey = 'order_id',
  preCombineField = 'ts'
);

主要建表属性说明如下表所示：

写入数据

DataSource API 方式：通过 df.write.format("hudi") 写入，关键写入参数（option）说明如下表所示：

支持的写入操作类型（hoodie.datasource.write.operation 取值）如下表所示：

DataSource API 写入示例（<路径> 为 Hudi 表存储根目录，表路径按 <路径>/<数据库>/<表名> 组织，实际取值以集群配置为准）：

inputDF.write.format("hudi") \
  .option("hoodie.datasource.write.table.name", "orders") \
  .option("hoodie.datasource.write.recordkey.field", "order_id") \
  .option("hoodie.datasource.write.partitionpath.field", "order_date") \
  .option("hoodie.datasource.write.precombine.field", "ts") \
  .option("hoodie.datasource.write.operation", "upsert") \
  .mode(Append) \
  .save("<路径>/hudi_db/orders")

Spark SQL 方式：使用 INSERT INTO 插入，MERGE INTO 实现 upsert（MERGE INTO 中 updates 为待合入数据的临时表，可先通过 CREATE TABLE ... USING parquet 建表并灌入数据）：

INSERT INTO hudi_db.orders VALUES
  ('1001', '2026-09-17', 99.0, 1760000000);

MERGE INTO hudi_db.orders AS t
USING updates AS s
ON t.order_id = s.order_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

UPDATE hudi_db.orders SET amount = 199.0 WHERE order_id = '1001';

DELETE FROM hudi_db.orders WHERE order_id = '1001';

查询数据

Hudi 表的查询类型由 hoodie.datasource.query.type 指定，默认 snapshot，各类型说明如下表所示：

增量查询需指定起始时刻（结束时刻可省略，默认为最新提交）：

spark.read.format("hudi") \
  .option("hoodie.datasource.query.type", "incremental") \
  .option("hoodie.datasource.read.begin.instanttime", "20260917093000123") \
  .load("<路径>/hudi_db/orders")