Apache Hudi — Incremental Data Processing, Record-Level Upserts, and CDC for Data Lakes

The Update Problem on a Data Lake

A raw object store is append-only by nature. Writing a new file is cheap; changing one row inside an existing Parquet file is not. For years the workaround was to rewrite whole partitions on every load, which turns a handful of changed records into gigabytes of churn.

Apache Hudi — pronounced “hoodie” — pioneered the transactional data lake to solve exactly this. It brings database primitives to files on cloud storage: tables, transactions, record-level upserts and deletes, indexes, and ingestion services, while keeping data in open formats readable by Spark, Flink, Presto, Trino, and Hive.

This guide walks the core model: the two table types, record keys and ordering fields, the write operations, indexing on the metadata table, the three query types, and the table services that keep a table healthy. If you are weighing formats first, our comparison of Delta Lake vs Iceberg vs Hudi is a useful companion.

Two Table Types: Copy-on-Write and Merge-on-Read

Every Hudi table is one of two types, and the choice shapes your write and read latency. Both organize data into file groups, each identified by a file ID and holding versioned file slices.

Copy-on-Write (CoW) stores data only in columnar base files (Parquet). An update rewrites the entire base file that contains the record. Reads are as fast as plain Parquet — zero read amplification — but writes pay for rewriting whole files even when few rows change.

Merge-on-Read (MoR) keeps a base file plus row-based log files (Avro) that capture incoming updates and deletes. Writes are cheap because changes are appended to the log; reads merge base and log at query time, so read cost grows with the number of changed records until a compaction reconciles them.

# Rule of thumb (from the Hudi docs' tradeoff table):
#
#              Copy-on-Write        Merge-on-Read
#  Write        higher latency       lower latency
#  Query        lower latency        higher latency
#  Update cost  rewrite base file    append to delta log
#  Write ampl.  O(file_groups)       O(records_changed)
#
# CoW  -> batch ETL, OLAP, slowly changing reference tables
# MoR  -> CDC, streaming ingestion, frequent updates/deletes

Record Keys and Ordering Fields

What makes an upsert possible is a stable identity per record. In Hudi that is the record key, set with hoodie.datasource.write.recordkey.field. When two versions of the same key arrive, the ordering field decides which one wins — typically an event timestamp.

The partition path (hoodie.datasource.write.partitionpath.field) controls physical layout. A key that is unique only within a partition uses a non-global index; a key that must be unique across the whole table uses a global index — more on that below.

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder.appName("hudi-demo")
    .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")
    .getOrCreate()
)

# Package coordinate (submit-time):
#   org.apache.hudi:hudi-spark3.5-bundle_2.13:1.2.0

hudi_options = {
    "hoodie.table.name": "trips",
    "hoodie.datasource.write.recordkey.field": "uuid",
    "hoodie.datasource.write.partitionpath.field": "city",
    "hoodie.table.ordering.fields": "ts",
    "hoodie.datasource.write.operation": "upsert",
}

(df.write.format("hudi")
    .options(**hudi_options)
    .mode("append")
    .save("s3a://lake/trips"))

The Spark session config above — the Kryo serializer, the HoodieCatalog, and the HoodieSparkSessionExtension — is the standard setup from the quick-start guide. The hudi-spark3.5-bundle_2.13 coordinate encodes the Spark and Scala versions; match them to your cluster.

The Write Operations

hoodie.datasource.write.operation selects how a write behaves. Picking the right one is the difference between a fast bootstrap and an accidental full-table merge.

• upsert — the default. Records are tagged as inserts or updates via an index lookup, so the table never shows duplicates. This is the CDC workhorse.
• insert — skips the index lookup, so it is faster but tolerates duplicates. Good for append-only logs.
• bulk_insert — a sort-based, disk-spilling write path that scales to hundreds of TBs. Use it for the initial load, not for incremental writes.
• delete — supports soft deletes (null the columns, keep the key) and hard deletes (drop the record entirely, or via a _hoodie_is_deleted column).
• insert_overwrite / insert_overwrite_table — replace the partitions present in the input, or the whole table, via a REPLACE_COMMIT. Ideal for backfills.
• delete_partition — bulk-drops whole partitions listed in hoodie.datasource.write.partitions.to.delete.

A common mistake is running upsert for a one-time historical load. Because upsert does an index lookup per record, it is far slower than bulk_insert for cold data — bootstrap with bulk_insert, then switch to upsert for the incremental stream.

Indexing: How Hudi Finds Records Fast

An upsert has to answer “where does this key already live?” before it can tag a record as an update. That is the index’s job, selected with hoodie.index.type. The default on Spark is SIMPLE, which joins incoming keys against keys read from storage.

BLOOM uses per-file bloom filters and key ranges to prune candidates. The GLOBAL_ variants enforce uniqueness across every partition — their lookup cost can scale with table size, whereas non-global indexes cost only O(records updated). BUCKET hashes keys into a fixed number of file groups for predictable, index-lookup-free writes.

The most impactful modern option is the Record Level Index (RLI). It lives in the metadata table — Hudi’s internal multi-modal index, itself a Hudi table under .hoodie/metadata. The metadata table also stores bloom-filter, column-stats, expression, and secondary-index partitions, so query engines can data-skip without opening base files.

# Enable the metadata table's record-level index for fast upserts
hoodie.metadata.enable = true
hoodie.metadata.record.level.index.enable = true
hoodie.index.type = RECORD_LEVEL_INDEX

# Or a global bloom index when keys must be unique table-wide
# hoodie.index.type = GLOBAL_BLOOM

Three Ways to Query

Hudi exposes the same table through different query types, chosen with hoodie.datasource.query.type.

Snapshot is the default: the latest committed state, merging base and log files on MoR so you always see the freshest data. Read-optimized (MoR only) reads base files alone for columnar-fast scans, at the cost of not seeing changes written since the last compaction.

Incremental is Hudi’s signature capability: give it a starting instant and it returns only the records that changed since, one per key. This is what lets you build efficient downstream pipelines that process deltas instead of rescanning the world — a natural fit for change data capture workflows.

# Snapshot query (latest merged state)
snapshot = spark.read.format("hudi").load("s3a://lake/trips")

# Incremental query: only rows changed since an instant
incr = (
    spark.read.format("hudi")
    .option("hoodie.datasource.query.type", "incremental")
    .option("hoodie.datasource.read.begin.instanttime", "20260715000000")
    .load("s3a://lake/trips")
)
incr.createOrReplaceTempView("trips_incr")
spark.sql("""
    SELECT _hoodie_commit_time, uuid, fare, ts
    FROM trips_incr
    WHERE fare > 20.0
""").show()

The _hoodie_commit_time column is one of several metadata columns Hudi maintains on every record. Time travel is the same idea in reverse: query the table as of a past instant to reproduce a report or debug a bad load.

Defining Tables with Spark SQL

You do not need the DataFrame API to create a Hudi table. Spark SQL supports USING hudi with table properties for the type, primary key, and ordering field. These properties are case-sensitive.

CREATE TABLE IF NOT EXISTS trips (
  uuid   STRING,
  rider  STRING,
  fare   DOUBLE,
  ts     BIGINT,
  city   STRING
) USING hudi
TBLPROPERTIES (
  type = 'mor',
  primaryKey = 'uuid',
  orderingFields = 'ts'
)
PARTITIONED BY (city);

-- Upserts are just MERGE / INSERT in SQL
MERGE INTO trips t
USING updates s
ON t.uuid = s.uuid
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

Set type = 'cow' for a Copy-on-Write table or type = 'mor' for Merge-on-Read. The MERGE INTO statement compiles down to the same upsert path as the DataFrame API, index lookup included.

Table Services: Compaction, Cleaning, Clustering

A Hudi table stays healthy through background table services. On MoR, compaction merges log files back into base files. It happens in two steps — scheduling writes a plan to the timeline, execution merges the file slices — and can run async (the default, non-blocking) or inline with the write.

# Inline compaction every 10 delta commits (Spark datasource / SQL)
hoodie.compact.inline = true
hoodie.compact.inline.max.delta.commits = 10
hoodie.compact.inline.trigger.strategy = NUM_COMMITS

# Or run it asynchronously (e.g. Spark Structured Streaming)
# hoodie.datasource.compaction.async.enable = true

Cleaning reclaims space by removing old file versions that no active query or incremental reader still needs, governed by a retention policy. Clustering rewrites data to improve layout — sorting by frequently filtered columns and coalescing small files — without changing record identity, which sharpens data-skipping for engines like Spark reading through predicate pushdown.

What Changed in Hudi 1.0

The 1.0 line, GA since January 2025, reframes Hudi as a database experience on the lake. Four changes matter most in practice.

• LSM tree timeline. The old active/archived split is gone; the timeline moved to .hoodie/timeline as a log-structured merge tree that scales to a very long history without slowing query planning.
• Non-blocking concurrency control (NBCC). Available for Flink streaming writers, it lets multiple streams write the same table concurrently without conflict resolution, requiring a lock only when committing to the timeline.
• Secondary and expression indexes. Index columns that are not the record key, or a function of a column, so filters on those columns get partition pruning and data skipping.
• Partial updates on MoR. Update a subset of columns without rewriting the whole record — cheaper writes for wide tables with narrow changes.

Production Checklist

• Pick the table type deliberately. CoW for read-heavy batch warehousing; MoR for CDC, streaming, and frequent deletes where you will run compaction.
• Bootstrap with bulk_insert. Load history once with bulk_insert, then switch the incremental stream to upsert.
• Enable the metadata table and RLI. The record-level index makes large-table upserts scale far better than a full-key join.
• Choose global vs non-global on purpose. Only pay for a global index when keys genuinely must be unique across all partitions.
• Tune compaction and cleaning. Match delta-commit triggers and retention to your read latency and incremental-reader needs; monitor log-file growth.
• Prefer incremental reads downstream. Consume changes with incremental queries instead of rescanning whole tables on every run.
• Test upgrades on a copy. Version-specific regressions exist; validate the writer and your key generator on staging before touching production.

Hudi’s bet is that a data lake should behave like a database for the operations that matter — upserts, deletes, and reading only what changed — without giving up open file formats or locking you into one engine. If your pipelines rewrite whole partitions to change a handful of rows, or your “CDC” is really a nightly full reload, modeling the table in Hudi is worth a spike.