Hudi

Hudi with Spark

Apache Hudi (pronounced “hoodie”) is the next-generation streaming data lake platform. Apache Hudi brings core warehouse and database functionality directly to a data lake. Hudi provides tables, transactions, efficient upserts/deletes, advanced indexes, streaming ingestion services, data clustering/compaction optimizations, and concurrency all while keeping your data in open-source file formats.

HUDI Spark3 Support Matrix

Hudi bundle compatibility by Spark version

Use the Hudi bundle that matches your Spark version. Using a bundle built for a different Spark version can cause class-loading issues, unexpected behavior, or Parquet write failures.

Hudi 1.2.0 is supported across all four Spark versions and has been validated with ODP 3.3.6.5.

Do not use a Hudi bundle intended for a different Spark version. Mismatched bundles can introduce classpath conflicts and affect other Spark operations.

Hudi 0.14.x with Spark 3.3.3

Older environments might still contain Hudi 0.14.x artifacts for Spark 3.3.3. If you must use Hudi 0.14.x, specify the fully qualified data source class:

org.apache.hudi.Spark32PlusDefaultSource

Hudi 0.14.x can register multiple data sources with the same short name. Hudi 1.2.0 resolves this issue, so you can use:

format("hudi")

Use Hudi 1.2.0 unless your environment specifically requires Hudi 0.14.x.

Use Hudi with Spark

Configure spark-shell or spark-submit

The existing spark-shell configuration also applies to Spark 4.1.1.

Use the Hudi bundle that matches your Spark and Scala versions:

/usr/odp/current/spark<line>-client/bin/spark-shell \ 
--jars hudi-spark<line>-bundle_<scala>-<version>.jar \ 
--conf spark.serializer=org.apache.spark.serializer.KryoSerializer \ 
--conf spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar \ 
--conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension \ 
--conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog

For Spark 4.1.1 environments that use Kerberos to access HDFS, also configure the required spark.driver.extraJavaOptions and spark.executor.extraJavaOptions settings for your Kerberos environment.

Auto-register a Hudi table in Hive Metastore from the Spark writer

On ODP 3.3.6.5, enable Hudi's built-in HMS sync in the Spark write and the table appears in default.<name> automatically, correctly wired to HoodieParquetInputFormat and the right partition schema — no separate beeline CREATE EXTERNAL TABLE step needed:

df.write.format("hudi")
 .option("hoodie.table.name", "my_hudi_table")
 .option("hoodie.datasource.write.recordkey.field", "id")
 .option("hoodie.datasource.write.precombine.field", "ts")
 .option("hoodie.datasource.write.partitionpath.field", "region")
 .option("hoodie.datasource.write.hive_style_partitioning", "true")
 .option("hoodie.datasource.write.table.type", "COPY_ON_WRITE")
 .option("hoodie.datasource.write.operation", "insert")
 .option("hoodie.datasource.hive_sync.enable", "true")
 .option("hoodie.datasource.hive_sync.mode", "hms")
 .option("hoodie.datasource.hive_sync.database", "default")
 .option("hoodie.datasource.hive_sync.table", "my_hudi_table")
 .option("hoodie.datasource.hive_sync.partition_fields", "region")
 .option("hoodie.datasource.hive_sync.partition_extractor_class",
 "org.apache.hudi.hive.MultiPartKeysValueExtractor")
 .option("hoodie.datasource.hive_sync.use_jdbc", "false")
 .mode("overwrite").save("hdfs:///path/to/my_hudi_table")

After the write completes, both routes read the same table:

• Spark: spark.sql("SELECT COUNT(*) FROM default.my_hudi_table") via spark_catalog — works on Spark 3.3.3, 3.5.1, 3.5.5, and 4.1.1.
• Beeline: SELECT COUNT(*) FROM default.my_hudi_table — works on Hive 4.1 provided the aux-lib jar from the "Hudi with Hive" section is in place.

use_jdbc=false routes the sync through the direct HMS thrift path. Leave this unless a cluster policy specifically requires HiveServer2 JDBC for DDL. auto_create_database=true is optional — add it if the target database might not exist.

Read and write Hudi data by path

The following example works across the supported Spark versions with Hudi 1.2.0:

val df = Seq(
 (1L, "us", 10.5),
 (2L, "eu", 22.0),
 (3L, "ap", 44.0)
).toDF("id", "region", "amount")
 
df.write
 .format("hudi")
 .option("hoodie.table.name", "my_hudi_table")
 .option("hoodie.datasource.write.recordkey.field", "id")
 .option("hoodie.datasource.write.partitionpath.field", "region")
 .option("hoodie.datasource.write.precombine.field", "amount")
 .option("hoodie.datasource.write.table.type", "COPY_ON_WRITE")
 .option("hoodie.datasource.write.hive_style_partitioning", "true")
 .mode("overwrite")
 .save("hdfs:///path/to/my_hudi_table")

Read the table by using the same HDFS path:

spark.read
  .format("hudi")
  .load("hdfs:///path/to/my_hudi_table")
  .show()

Use the hudi data source

With Hudi 1.2.0, use the standard hudi data source for both read and write operations:

.format("hudi")

This syntax is supported with Hudi 1.2.0 across Spark 3.3.3, Spark 3.5.1, Spark 3.5.5, and Spark 4.1.1.

Older Hudi 0.14.x bundles on Spark 3.3.3 can return the following error:

spark.read
  .format("org.apache.hudi.Spark32PlusDefaultSource")
  .load("hdfs:///path/to/my_hudi_table")
  .show()

If you must use Hudi 0.14.x, specify the fully qualified data source class for both read and write operations:

.format("org.apache.hudi.Spark32PlusDefaultSource")

For supported ODP deployments, use Hudi 1.2.0 and the bundle that corresponds to your Spark version.

Import Statements

This code block imports necessary libraries for Spark and Hudi operations, and it initializes variables for the table name and base path to be used in subsequent Hudi data operations.

// spark-shell
import scala.collection.JavaConversions._
import org.apache.spark.sql.SaveMode._
import org.apache.hudi.DataSourceReadOptions._
import org.apache.hudi.DataSourceWriteOptions._
import org.apache.hudi.common.table.HoodieTableConfig._
import org.apache.hudi.config.HoodieWriteConfig._
import org.apache.hudi.keygen.constant.KeyGeneratorOptions._
import org.apache.hudi.common.model.HoodieRecord
import spark.implicits._
val tableName = "hudi_table"
val basePath = "hdfs:///warehouse/tablespace/external/hive/hudi_table"

Create Table, Insert Data, and Query Data

This code block demonstrates how to create a Hudi table, insert data into it, and then query that data using Spark SQL, showcasing a complete cycle of table creation and data manipulation.

spark.sql("CREATE TABLE if not exists hudi_table (     ts BIGINT,     uuid STRING,     rider STRING,     driver STRING,     fare DOUBLE,     city STRING ) USING HUDI PARTITIONED BY (city)")
val columns = Seq("ts","uuid","rider","driver","fare","city")
val data =
 Seq((1695159649087L,"334e26e9-8355-45cc-97c6-c31daf0df330","rider-A","driver-K",19.10,"san_francisco"),
 (1695091554788L,"e96c4396-3fad-413a-a942-4cb36106d721","rider-C","driver-M",27.70 ,"san_francisco"),
 (1695046462179L,"9909a8b1-2d15-4d3d-8ec9-efc48c536a00","rider-D","driver-L",33.90 ,"san_francisco"),
 (1695516137016L,"e3cf430c-889d-4015-bc98-59bdce1e530c","rider-F","driver-P",34.15,"sao_paulo" ),
 (1695115999911L,"c8abbe79-8d89-47ea-b4ce-4d224bae5bfa","rider-J","driver-T",17.85,"chennai"));
var inserts = spark.createDataFrame(data).toDF(columns:_*)
inserts.write.format("hudi").
 option(PARTITIONPATH_FIELD_NAME.key(), "city").
 option(TABLE_NAME, tableName).
 mode(Overwrite).
 save(basePath)
val tripsDF = spark.read.format("hudi").load(basePath)
tripsDF.createOrReplaceTempView("trips_table")
spark.sql("SELECT uuid, fare, ts, rider, driver, city FROM  trips_table WHERE fare > 20.0").show()
spark.sql("SELECT _hoodie_commit_time, _hoodie_record_key, _hoodie_partition_path, rider, driver, fare FROM  trips_table").show()

Update Data and Read Data

This code block reads data from a Hudi table, modifies the 'fare' column for a specific rider, and updates the table with the new information.

// Lets read data from target Hudi table, modify fare column for rider-D and update it. 
val updatesDf = spark.read.format("hudi").load(basePath).filter($"rider" === "rider-D").withColumn("fare", col("fare") * 10)
updatesDf.write.format("hudi").
 option(OPERATION_OPT_KEY, "upsert").
 option(PARTITIONPATH_FIELD_NAME.key(), "city").
 option(TABLE_NAME, tableName).
 mode(Append).
 save(basePath)
spark.sql("SELECT _hoodie_commit_time, _hoodie_record_key, _hoodie_partition_path, rider, driver, fare FROM  hudi_table").show()

Merging Data and Read Data

This code block demonstrates how to merge data from a source Hudi table into a target Hudi table, illustrating the integration of datasets within the Hudi framework.

-- source table using Hudi for testing merging into target Hudi table
spark.sql("CREATE TABLE fare_adjustment (ts BIGINT, uuid STRING, rider STRING, driver STRING, fare DOUBLE, city STRING) USING HUDI")
spark.sql("INSERT INTO fare_adjustment VALUES (1695091554788,'e96c4396-3fad-413a-a942-4cb36106d721','rider-C','driver-M',-2.70 ,'san_francisco'),(1695530237068,'3f3d9565-7261-40e6-9b39-b8aa784f95e2','rider-K','driver-U',64.20 ,'san_francisco'),(1695241330902,'ea4c36ff-2069-4148-9927-ef8c1a5abd24','rider-H','driver-R',66.60 ,'sao_paulo'    ), (1695115999911,'c8abbe79-8d89-47ea-b4ce-4d224bae5bfa','rider-J','driver-T',1.85,'chennai'))"
spark.sql("MERGE INTO hudi_table AS target USING fare_adjustment AS source ON target.uuid = source.uuid WHEN MATCHED THEN UPDATE SET target.fare = target.fare + source.fare WHEN NOT MATCHED THEN INSERT * ")

Delete Data

This code block loads data from a Hudi table, filters out records corresponding to a specific rider, and prepares these records for deletion from the table.

val deletesDF = spark.read.format("hudi").load(basePath).filter($"rider" === "rider-F")
deletesDF.write.format("hudi").
 option(OPERATION_OPT_KEY, "delete").
 option(PARTITIONPATH_FIELD_NAME.key(), "city").
 option(TABLE_NAME, tableName).
 mode(Append).
 save(basePath)

Time Travel Query

Time travel queries in Hudi allow you to view and query data as it appeared at specific points in time, using different timestamp formats to access historical data snapshots.

spark.read.format("hudi").
 option("as.of.instant", "20210728141108100").
 load(basePath)
spark.read.format("hudi").
 option("as.of.instant", "2021-07-28 14:11:08.200").
 load(basePath)
// It is equal to "as.of.instant = 2021-07-28 00:00:00"
spark.read.format("hudi").
 option("as.of.instant", "2021-07-28").
 load(basePath)
// Same example in Spark SQL
-- time travel based on commit time, for eg: `20220307091628793`
SELECT * FROM hudi_table TIMESTAMP AS OF '20220307091628793' WHERE id = 1;
-- time travel based on different timestamp formats
SELECT * FROM hudi_table TIMESTAMP AS OF '2022-03-07 09:16:28.100' WHERE id = 1;
SELECT * FROM hudi_table TIMESTAMP AS OF '2022-03-08' WHERE id = 1;

Capture the Data Change Query

Hudi also exposes first-class support for Change Data Capture (CDC) queries. CDC queries are useful for applications that need to obtain all the changes, along with before or after images of records, given a commit time range.

// spark-shell
// Lets first insert data to a new table with cdc enabled.
val columns = Seq("ts","uuid","rider","driver","fare","city")
val data =
 Seq((1695158649187L,"334e26e9-8355-45cc-97c6-c31daf0df330","rider-A","driver-K",19.10,"san_francisco"),
 (1695091544288L,"e96c4396-3fad-413a-a942-4cb36106d721","rider-B","driver-L",27.70 ,"san_paulo"),
 (1695046452379L,"9909a8b1-2d15-4d3d-8ec9-efc48c536a00","rider-C","driver-M",33.90 ,"san_francisco"),
 (1695332056404L,"1dced545-862b-4ceb-8b43-d2a568f6616b","rider-D","driver-N",93.50,"chennai"));
var df = spark.createDataFrame(data).toDF(columns:_*)
// Insert data
df.write.format("hudi").
 option(PARTITIONPATH_FIELD_NAME.key(), "city").
 option(CDC_ENABLED.key(), "true").
 option(TABLE_NAME, tableName).
 mode(Overwrite).
 save(basePath)
// Update fare for riders: rider-A and rider-B 
val updatesDf = spark.read.format("hudi").load(basePath).filter($"rider" === "rider-A" || $"rider" === "rider-B").withColumn("fare", col("fare") * 10)
updatesDf.write.format("hudi").
 option(OPERATION_OPT_KEY, "upsert").
 option(PARTITIONPATH_FIELD_NAME.key(), "city").
 option(CDC_ENABLED.key(), "true").
 option(TABLE_NAME, tableName).
 mode(Append).
 save(basePath)
// Query CDC data
spark.read.option(BEGIN_INSTANTTIME.key(), 0).
 option(QUERY_TYPE.key(), QUERY_TYPE_INCREMENTAL_OPT_VAL).
 option(INCREMENTAL_FORMAT.key(), "cdc").
 format("hudi").load(basePath).show(false)

For more details, see Apache Quick Start Guide.

Hudi with Hive

Use the Hudi Hadoop MR bundle for Hive's auxiliary JAR path. Do not use the Hudi Spark bundle.

Hive Configuration

1. Update the Hive Environment Configuration:
   • In the Ambari UI, go to Hive > Configs.
   • In hive-env.sh, add the Hudi Hadoop MR bundle to HIVE_AUX_JARS_PATH:

export export HIVE_AUX_JARS_PATH=${HIVE_AUX_JARS_PATH}:/usr/odp/current/hudi/hudi-hadoop-mr-bundle-1.2.0.jar

1. Configure the Hive Properties:
   • Update the hive-site.xml file with the following properties:

<property>
 <name>hive.input.format</name>
 <value>org.apache.hudi.hadoop.HoodieParquetInputFormat</value>
</property>
<property>
 <name>hive.tez.input.format</name>
 <value>org.apache.hadoop.hive.ql.io.HiveInputFormat</value>
</property>

Do NOT sethive.tez.input.format=org.apache.hudi.hadoop.hive.HoodieCombineHiveInputFormat. Some older Hudi guides for Hive 3.x recommend that combine wrapper; on Hive 4.1 it hard-fails every SELECT with:

java.lang.NoSuchMethodError: 'void org.apache.hadoop.hive.ql.log.PerfLogger.PerfLogBegin(java.lang.String, java.lang.String)'
  at org.apache.hudi.hadoop.hive.HoodieCombineHiveInputFormat.getSplits(HoodieCombineHiveInputFormat.java:326)
 

Hive 4 removed the PerfLogger.PerfLogBegin(String, String) signature that Hudi's Hive-3-era combine wrapper calls. Leave hive.tez.input.format at the default HiveInputFormat shown above — Hudi's per-table HoodieParquetInputFormat from the CREATE TABLE DDL still takes effect.

1. Restart the Hive Services:
   • Restart the Hive services to apply the changes.
2. Create Hudi Tables:

Create Hudi Tables

Create Hudi COW Table

CREATE EXTERNAL TABLE hudi_cow_table_new (
 `_hoodie_commit_time` STRING,
 `_hoodie_commit_seqno` STRING,
 `_hoodie_record_key` STRING,
 `_hoodie_partition_path` STRING,
 `_hoodie_file_name` STRING,
 id STRING,
 name STRING,
 age INT,
 ts TIMESTAMP
)
PARTITIONED BY (dt STRING)
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT 'org.apache.hudi.hadoop.HoodieParquetInputFormat'
 OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION 'hdfs:///tmp/hudi/hudi_cow_table_new';
 
MSCK REPAIR TABLE hudi_cow_table_new;
SELECT COUNT(*) FROM hudi_cow_table_new;

• Backtick-quote every _hoodie_* meta column. Hive 4.1's parser rejects unquoted identifiers that begin with _ and fails at DDL parse time. (The Realtime example already does this; the CoW/MoR examples don't.)
• MSCK REPAIR TABLE only picks up partitions when the Hudi write set hoodie.datasource.write.hive_style_partitioning=true (already documented in the Spark section). Without that option, MSCK REPAIR fails with MetastoreException: Invalid partition name hdfs:///path/to/hudi_table/<partition-value>. Hudi has no offline operation to rename partition directories, so the table has to be rewritten.
• Hudi table properties like type, primaryKey, preCombineField belong on the Spark writer side (hoodie.datasource.write.* options), not in Hive's TBLPROPERTIES. They're set once by the Spark job that authored the table and stored in Hudi's own .hoodie/ metadata directory.
• Prefer the auto-sync path — enabling hoodie.datasource.hive_sync.mode=hms on the Spark writer registers the table in Hive Metastore automatically with the correct InputFormat, partition schema, and meta columns. No hand-written CREATE EXTERNAL TABLE step needed.

Insert Data into Hudi COW Table

INSERT INTO hudi_cow_table_new
VALUES ('1', 'Alice', 30, '2024-01-01 00:00:00', '2024-01-01');
INSERT INTO hudi_cow_table_new
VALUES ('1', 'Alice', 31, '2024-01-02 00:00:00', '2024-01-01');

Create Hudi MOR Table

-- MoR read-optimised view (base parquet only)
CREATE EXTERNAL TABLE hudi_mor_table_new_ro (
 `_hoodie_commit_time` STRING,
 `_hoodie_commit_seqno` STRING,
 `_hoodie_record_key` STRING,
 `_hoodie_partition_path` STRING,
 `_hoodie_file_name` STRING,
 id STRING,
 name STRING,
 age INT,
 ts TIMESTAMP
)
PARTITIONED BY (dt STRING)
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT 'org.apache.hudi.hadoop.HoodieParquetInputFormat'
 OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION 'hdfs:///tmp/hudi/hudi_mor_table_new';
 
-- MoR realtime view (base parquet + log files merged at read time)
CREATE EXTERNAL TABLE hudi_mor_table_new_rt (
 `_hoodie_commit_time` STRING,
 `_hoodie_commit_seqno` STRING,
 `_hoodie_record_key` STRING,
 `_hoodie_partition_path` STRING,
 `_hoodie_file_name` STRING,
 id STRING,
 name STRING,
 age INT,
 ts TIMESTAMP
)
PARTITIONED BY (dt STRING)
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT 'org.apache.hudi.hadoop.realtime.HoodieParquetRealtimeInputFormat'
 OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION 'hdfs:///tmp/hudi/hudi_mor_table_new';

Insert Data into Hudi MOR Table

INSERT INTO hudi_mor_table_new
VALUES ('1', 'Alice', 30, '2024-06-07 14:00:00', '2024-06-07');

Create Hudi Realtime Table

CREATE EXTERNAL TABLE hudi_test_realtime_new(
 `_hoodie_commit_time` STRING,
 `_hoodie_record_key` STRING,
 `waybillid` STRING,
 `quantity` INT,
 `createtm` STRING)
PARTITIONED BY (
 `sourcezonecode` STRING)
ROW FORMAT SERDE
 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT
 'org.apache.hudi.hadoop.realtime.HoodieParquetRealtimeInputFormat'
OUTPUTFORMAT
 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
 'hdfs:///tmp/hudi/hudi_table_realtime_new'
TBLPROPERTIES (
 'last_commit_time_sync' = '20190430150335',
 'transient_lastDdlTime' = 'CURRENT_TIMESTAMP');

Insert Data into Hudi Realtime Table

INSERT INTO hudi_test_realtime_new
VALUES
('2024-06-07 14:00:00', 'record_key_1', 'waybill_1', 10, '2024-06-07 14:05:00', 'source_zone_1'),
('2024-06-07 14:00:00', 'record_key_2', 'waybill_2', 15, '2024-06-07 14:10:00', 'source_zone_2');

Perform Time Travel Queries

For time travel queries in Hudi, Hive does not natively support querying historical versions. Instead, you should use Spark for advanced time travel capabilities. However, you can still perform some operations to query the latest data or specific snapshots.

Query Latest Data (Hive)

SELECT * FROM hudi_cow_table_new WHERE dt = '2024-01-01';
SELECT * FROM hudi_mor_table_new WHERE dt = '2024-06-07';

Query Specific Snapshot (Using Spark)

If you need to query historical snapshots, use Spark to read from specific commit times or timestamps:

val df = spark.read.format("hudi")
 .option("hoodie.snapshot.timestamp", "2024-01-01T00:00:00.000Z") // Example timestamp
 .load("hdfs:///tmp/hudi/hudi_cow_table_new")
df.show()

Cross-Spark-version interop

Hudi tables in ODP 3.3.6.5 round-trip cleanly across every Spark line on a shared HDFS location, provided each Spark job loads the matching Hudi bundle for its Spark version:

Users do not need to standardise on a "lowest common denominator" Spark version. Mixed writers and readers on the same Hudi lake work in every direction.