Deep Dive into Apache Hudi

Apache Hudi (Hadoop Upserts Deletes and Incrementals) is an open-source transactional data lakehouse framework that brings database-like ACID capabilities, Change Data Capture (CDC), record-level upserts/deletes, and incremental processing directly to cloud blob storage (such as AWS S3, Google Cloud Storage, Azure ADLS) and HDFS.

In this chapter, we explore the problems Hudi solves, its mechanical architecture, how data is stored on disk, how it tracks incremental changes, and walk through a step-by-step example of inserts, updates, and deletes.

1. What Problem Does Apache Hudi Solve?

Traditional data lake file formats (such as raw Parquet, ORC, or CSV) are static and designed for append-only, batch data workloads. When organizations attempted to build analytical architectures directly on raw files, they hit several fundamental bottlenecks:

• The "All-or-Nothing" Rewrite Problem (High Write Amplification): If a user requested a delete due to compliance, or if a single customer's details were updated, there was no way to modify a single row. The execution engine had to read a partition of files (potentially millions of rows), modify or remove the target record in memory, and rewrite a completely new set of files. For high-velocity streaming data, this caused massive write amplification and cost.
• Lack of ACID Transactions (Inconsistent Queries): Raw cloud storage has no transactional capabilities. If a Spark job writing new files failed halfway, readers would query a mixture of old and partially written new files, leading to dirty reads, duplicates, or corrupted data.
• The Small File Problem: Continuous streaming engines (like Kafka or Flink) ingestion models write data in micro-batches. If they write to Parquet directly every few minutes, they generate thousands of tiny, KB-sized files. Over time, this overwhelms file system metadata catalogs (such as HDFS NameNodes or cloud storage API rate limits) and slows query times to a crawl.
• The Ingestion Latency Trade-off: Organizations had to choose between query speed and data freshness. They could either query raw JSON/CSV files in real-time (very slow query speed) or wait for a daily batch job to partition and convert raw files into Parquet (slow data freshness).

2. How Are These Problems Solved?

Apache Hudi inserts a transactional storage management layer between query engines and raw file storage. It resolves the limitations of raw formats through these mechanisms:

• Record-Level Operations: Hudi assigns unique keys to every record. Updates and deletes are routed directly to the specific file groups holding those keys, avoiding the need to rewrite unrelated files.
• Active Commit Timeline (ACID compliance): Hudi manages an active, transaction-backed log (the timeline) under the .hoodie/ directory. Reads are isolated from writes. Query engines only see completed, committed states, while inflight transactions are completely hidden. If a job fails, Hudi uses the timeline to identify and roll back incomplete files.
• Dynamic Write-Time File Clustering & Bin-Packing: During ingestion, Hudi constructs a workload profile of target partitions. It detects existing files that are smaller than a configured limit (e.g. 100MB) and packs new incoming records directly into those files first (Auto-Bucket sizing) rather than creating new ones. Out-of-band clustering jobs can also reorganize small files in the background.
• Decoupled Compute and Storage views (COW vs. MOR): Hudi enables the Lakehouse model by letting users choose how updates are written: either immediately rewritten into columnar base files (Copy on Write) or appended quickly to row-based delta logs (Merge on Read), decoupling read performance from write speed.

3. Core Architecture & Concepts

Apache Hudi acts as a storage layer management system that organizes files on disk while maintaining metadata. It leverages three primary core primitives to uniquely identify and order data records:

• Record Key: Uniquely identifies a record within a partition or table. It is configured via hoodie.datasource.write.recordkey.field.
• Partition Path: Defines the physical directory layout on storage under which records are stored. Configured via hoodie.datasource.write.partitionpath.field.
• Precombine Key: Resolves ordering conflicts during upsert operations. Configured via hoodie.datasource.write.precombine.field. When multiple records with the same record key arrive in the same write batch, Hudi keeps the record containing the largest precombine value.

4. Hudi Table Types

Hudi categorizes storage layouts into two table types, balancing read performance against write latency:

• Copy on Write (COW):
  • Data Layout: Stored strictly as columnar files (Apache Parquet).
  • Write Mechanism: Every write/update operation rewrites the entire base Parquet file containing the updated or newly inserted records.
  • Trade-off: High write amplification and higher write latency, but provides maximum read performance with zero read-side merging overhead.
  • Best For: Read-heavy analytical tables with low update frequencies.
• Merge on Read (MOR):
  • Data Layout: Hybrid layout consisting of columnar base files (Apache Parquet) and row-based append-only delta logs (Apache Avro).
  • Write Mechanism: Updates are appended directly to Avro log files without rewriting base files immediately.
  • Compaction: An asynchronous or inline background process merges Avro log files back into new Parquet base files to prevent infinite log growth.
  • Trade-off: Extremely low write latency and minimal write amplification, but query latency varies depending on the volume of uncompacted delta logs.
  • Best For: High-frequency CDC feeds, streaming ingestion (Kafka, Flink, Spark Streaming), and real-time data pipelines.

5. Query Views & Types

Depending on the table type and view required, Hudi supports three query views:

6. How Hudi Tracks Incremental Data

Hudi maintains an internal metadata architecture designed to locate records instantly and read only the changes (deltas) between specific periods of time.

Incoming Record
  ├── key: R1
  └── partition: 2026
        │
        v
    Hudi Index (e.g., Record-Level Index)
        │
        v
    Maps R1 -> File Group A (FileID_A)
        │
        +------------------------------------------+
        |                                          |
        v (COW Table)                              v (MOR Table)
    Rewrite FileGroup A base file              Append R1 to FileGroup A
    to include modified record.                delta log: FileID_A_t1_t2.log

• Record Keys & Metadata Columns: Every record is written with a unique key. When Hudi writes a record, it automatically injects five metadata columns into the row: _hoodie_commit_time (timestamp of transaction), _hoodie_record_key (unique ID), _hoodie_partition_path (partition directory), _hoodie_file_name (physical file location), and _hoodie_commit_seqno (ordering sequence number).
• File Groups and Slices: Data files are grouped logically into File Groups. A File Group is identified by a unique File ID and contains all versions/slices of records mapping to that group. Inside a File Group, data is organized into File Slices. A File Slice contains a single columnar base file (.parquet) and zero or more delta logs (.log containing Avro/JSON updates) tied to a specific commit time.
• Indexing Mapping: When a batch of updates arrives, Hudi uses an Index (such as a Bloom filter stored in the Parquet footer or the high-performance point-lookup Record-Level Index) to map the incoming record key to its existing File ID (File Group location). If found, the update is routed to that File Group. If not, it is marked as a new insert.
• Incremental Timeline Pruning: For downstream consumers querying incrementals, Hudi scans the timeline commits between beginInstantTime and endInstantTime. The query reader bypasses all unmodified file groups, reading only the base Parquet files and delta logs written or modified within that transaction window.

7. How Hudi Data is Stored physically

Hudi organizes directories on storage using a standard database-style partition layout, accompanied by an active metadata transaction log:

table_root_directory/
  ├── .hoodie/                           <-- Timeline & Transaction logs
  │     ├── t1.commit                    <-- Completed commit metadata
  │     ├── t2.deltacommit.inflight      <-- Inflight MOR update transaction
  │     └── metadata/                    <-- Internal Metadata Table (HFiles)
  │
  ├── year=2026/                         <-- Partition Folder
  │     ├── partition_metadata           <-- Hudi partition marker
  │     ├── FileID_A_t1.parquet          <-- Base Parquet file (Slice 1)
  │     ├── .FileID_A_t1_t2.log.1        <-- Avro delta log file (updates at t2)
  │     ├── FileID_B_t2.parquet          <-- Base Parquet file (Slice 1, Group B)

1. The .hoodie/ directory: Stores the timeline commits. These files (e.g. .commit, .deltacommit, .cleaner, .compaction) are small JSON files containing metadata about which physical files were created, updated, or removed, along with statistics like total records written.
2. Base columnar files (.parquet): The actual data stored in columnar format. In a COW table, these represent the absolute latest state of the table. In a MOR table, they represent the latest compacted state.
3. Delta log files (.log.<sequence_number>): Appended directly to storage in MOR tables. These are binary row-oriented files (usually written using Apache Avro format) containing row-level inserts, updates, and delete tombstone blocks.

8. Indexing Strategies

Indexes map incoming record keys to existing File IDs to avoid scanning entire partitions during upsert operations.

• Bloom Index (Default for Spark): Uses Bloom filters stored in Parquet file footers to prune files during key lookups.
• Global Bloom Index: Checks keys across the entire table (all partitions) rather than just target partitions.
• Simple / Global Simple Index: Performs a join between incoming record keys and key extraction from physical data files.
• HBase / Record-Level Index (RLI): Secondary index storing key-to-location mappings in an external storage layer or internal Hudi Metadata Table for low-latency lookups.
• Hudi Metadata Table: An internal multi-modal index (built as a Hudi MOR table) that stores file listings, column stats, and bloom filters to eliminate expensive file system listing operations (LIST calls on object storage).

9. Table Operations & Optimization

• Compaction (MOR only): Configured via inline or asynchronous schedules (e.g., hoodie.compact.inline.max.delta.commits). Compaction merges Avro delta logs with Parquet base files to form new consolidated base files.
• Clustering: Groups small files into optimal size blocks (128 MB - 512 MB) and optionally applies space-filling curves (Z-order or Hilbert) on query predicate columns to accelerate search.
• Concurrency Control: Supports Multi-Version Concurrency Control (MVCC) for readers and writers, alongside Optimistic Concurrency Control (OCC) using external lock providers (Zookeeper, Hive Metastore, DynamoDB) for concurrent writers.

10. Step-by-Step Example: Record Lifecycle (MOR Table)

Let's walk through a physical step-by-step example of how Hudi processes inserts, updates, and deletes under the hood on a Merge on Read (MOR) table across a series of commit instants (t1, t2, t3, t4).

State 0: Initial Inserts at Instant t1

• Operation: Insert three records: R1 (value: "Alice"), R2 (value: "Bob"), and R3 (value: "Charlie") into partition year=2026.

Hudi Action:

1. Hudi checks the index and determines these are new inserts.
2. Hudi creates a new File Group A (assigns it FileID_A).
3. It writes these records into a new base Parquet file FileID_A_t1.parquet (containing R1, R2, R3 with _hoodie_commit_time set to t1).
4. It writes t1.commit to .hoodie/ to declare the transaction completed.

State 1: Updates & Inserts at Instant t2

• Operation: Update the value of R2 to "Robert" and insert a new record R4 (value: "David", partition: year=2026).

Hudi Action:

1. Hudi index lookup maps R2 -> File Group A (match), and R4 -> New insert (no match).
2. Hudi writes the new insert R4 to a new File Group B base file FileID_B_t2.parquet.
3. Hudi writes the update R2 by appending a row to a delta log in File Group A: .FileID_A_t1_t2.log.1 (contains the updated row R2: "Robert" with commit time t2).
4. It writes t2.deltacommit to .hoodie/.

ℹ️ NOTE: Reading at State 1:

• Snapshot Query: Reads FileID_A_t1.parquet and merges it in memory with .FileID_A_t1_t2.log.1. It returns: R1 ("Alice"), R2 ("Robert"), R3 ("Charlie"), and R4 ("David").
• Read Optimized Query: Reads only base files: FileID_A_t1.parquet and FileID_B_t2.parquet. It returns: R1 ("Alice"), R2 ("Bob"), R3 ("Charlie"), and R4 ("David") (the update is omitted since logs are not parsed).

State 2: Hard Delete at Instant t3

• Operation: Delete record R1.

Hudi Action:

1. Hudi index lookup maps R1 -> File Group A.
2. Hudi appends a HoodieDeleteBlock containing the record key of R1 to File Group A's active delta log: .FileID_A_t1_t2.log.1 (appended as block sequence, or a new log version .FileID_A_t1_t3.log.2).
3. It writes t3.deltacommit to .hoodie/.

State 3: Asynchronous Compaction at Instant t4

• Operation: Rebuild base files to consolidate logs (typically triggered when log volume or commit counts exceed limits).

Hudi Action:

1. A background compaction process reads the original base file FileID_A_t1.parquet and all its delta logs.
2. It applies the updates (overwriting R2 to "Robert") and deletes (removing R1).
3. It writes a brand new consolidated base Parquet file for File Group A: FileID_A_t4.parquet (containing only R2: "Robert" and R3: "Charlie").
4. The old base file FileID_A_t1.parquet and its logs are marked as obsolete, to be purged by the next Hudi Cleaner run.
5. It writes t4.compaction to .hoodie/.

11. Transactions, Commits, and Concurrency

Apache Hudi handles transactions, atomic commits, and isolation guarantees using three main components: a centralized Timeline, Multi-Version Concurrency Control (MVCC), and Optimistic/Non-Blocking Concurrency Control.

11.1 The Core Engine: The Hudi Timeline (Write-Ahead Log)

At the heart of Hudi’s transaction model is the Timeline. Located inside the .hoodie/ directory on cloud object storage (S3/GCS/HDFS), the timeline acts as an append-only Write-Ahead Log (WAL) containing a series of Instants.

Every write or table maintenance operation follows a strict 3-state transition on the timeline:

1. REQUESTED: The write action is planned (e.g., 20260812000000.commit.requested).
2. INFLIGHT: The write executors are actively writing data files or log files (20260812000000.commit.inflight).
3. COMPLETED: The write is finalized and atomically published (20260812000000.commit).

ℹ️ NOTE: Snapshot Isolation for Readers: Queries scan the Timeline and only read data associated with COMPLETED instants. Uncommitted (INFLIGHT) data files are completely invisible to readers.

11.2 Low-Level Storage: Multi-Version Concurrency Control (MVCC)

Hudi provides MVCC between reader queries, ingestion writers, and background table services (compaction, clustering, cleaning):

• Append-Only Immutable Files: Hudi never mutates base Parquet files in place. Instead, it writes new base files (for Copy-on-Write) or appends delta log files (for Merge-on-Read) tagged with unique File IDs and Instant Timestamps.
• Lock-Free Reads: Readers query a point-in-time snapshot of file slices without acquiring locks or blocking writers.
• Decoupled Table Services: Asynchronous background jobs—like compaction (merging log files into base Parquet files) or clustering (re-organizing layout)—run concurrently using MVCC without holding up streaming ingested writes.

11.3 Concurrency Control Across Multiple Writers

When multiple distributed writers (e.g., Spark, Flink, Kafka Streamers) attempt to write to the same Hudi table at the same time, Hudi supports two primary concurrency modes:

Mode A: Optimistic Concurrency Control (OCC)

OCC assumes write-write conflicts are rare. Writers operate independently and acquire short-lived distributed locks only during critical metadata operations.

How the OCC Transaction Lifecycle Works:

1. Start Instant: Writer creates an INFLIGHT instant on the Timeline stamped with start time T₁.
2. Write Payload: Data is written to storage (new Parquet files or log files).
3. Acquire Distributed Lock: Right before committing, the writer acquires a short lock (via DynamoDB, ZooKeeper, Hive Metastore, or S3/FileSystem-based locks).
4. Conflict Check: The writer checks the Timeline for any commits completed between T₁ and now that modified the exact same File Groups (partitions + file IDs).
   • No Conflict: The transaction completes. The writer commits the metadata file (.commit), updates the Hudi Metadata Table, and releases the lock.
   • Conflict Detected: The writer rolls back its current attempt, marks the instant as failed, and optionally retries based on configured policies.
5. Release Lock: The lock is held for mere milliseconds to update metadata, keeping lock contention low.

Mode B: Non-Blocking Concurrency Control (NBCC)

For high-frequency multi-writer ingestion (e.g., multiple streaming pipelines appending to the same partitions), OCC locks can lead to writer starvation. Hudi provides Non-Blocking Concurrency Control (NBCC):

• Lockless Log Appends: On Merge-On-Read (MOR) tables, concurrent writers append their updates to separate log files under the same file group without acquiring locks or aborting.
• Deferred Conflict Resolution: Conflict resolution is deferred to the Compactor or Reader. During compaction, Hudi uses the globally ordered completion timestamps and pre-combine keys (e.g., event timestamps) to merge record payloads deterministically.

11.4 Metadata Table & Atomic File Listing

Listing files on cloud object stores (like S3) is notoriously slow and eventually consistent. Hudi solves this with an internal Hudi Metadata Table (an embedded, indexed Merge-On-Read table inside .hoodie/metadata).

• Single Source of Truth: Instead of scanning S3 file trees, readers query the Metadata Table to resolve valid data files.
• Transactional Updates: Updates to the Metadata Table are committed atomically alongside timeline instants as part of the overall transaction.

Technical Q&A

Q1: How does Hudi handle concurrent writes, and what happens when two writers attempt to update the same record simultaneously?

Answer: Hudi uses a combination of Multi-Version Concurrency Control (MVCC) for reader-writer isolation and Optimistic Concurrency Control (OCC) for writer-writer conflict resolution.

• Reader-Writer Isolation: Readers only see completed instants on the Hudi timeline. Inflight operations (such as active deltacommit or compaction steps) write to isolated file slices or append new delta blocks without locking out readers.
• Writer-Writer Conflicts: When two concurrent streaming jobs or batch ingestion pipelines attempt to write to the same table:
  • Locking Provider: Writers acquire a lightweight lock (via Zookeeper, Hive Metastore, or DynamoDB) only during the timeline commit/validation phase.
  • Conflict Detection: Hudi checks if the file slices modified by the incoming commit overlap with any instant completed after the incoming writer started its ingestion task.
  • Resolution: If the exact same file group was modified by another writer during that window, Hudi throws a ConcurrentModificationException and rolls back the conflicting inflight commit.
• Payload-Level Merging: If the target records reside in different file groups, both writes succeed without conflict. If fine-grained record-level concurrent updates are required without aborts, writers must route through a centralized ingestion queue (e.g., Kafka) or use custom HoodieMerger classes that handle delta-log merging asynchronously.

Q2: Why does Hudi's Bloom Index struggle at multi-terabyte/petabyte scales with high key cardinality, and how does the Record-Level Index (RLI) resolve this?

Answer:

• Bloom Index Bottleneck: The standard Bloom Index stores a Bloom filter inside the footer of every Parquet base file. During an upsert batch, Hudi must read the metadata/footer of target files to evaluate key membership. With billions of records spread across hundreds of thousands of files, false positives force actual index lookups against Parquet files, creating severe I/O bottlenecks.
• Global Bloom Index: Lookup cost scales at O(K × G), where K is the number of incoming keys and G is the total number of file groups.
• Record-Level Index (RLI) Resolution: RLI completely bypasses file listing and footer scans by maintaining a point-lookup key-to-location mapping (RecordKey -> FileID) inside Hudi's internal Metadata Table (which itself is an indexed MOR table). Lookups become O(K) point lookups against an internal HFile format within the Metadata Table. It delivers consistent sub-second key locations regardless of dataset size or partition depth, eliminating object store LIST operations entirely.

Q3: In a high-churn CDC ingestion pipeline on a MOR table, how do you balance read amplification against write amplification?

Answer: The trade-off manifests as Log Accumulation (Low Write Amplification / High Read Amplification) vs. Aggressive Compaction (High Write Amplification / Low Read Amplification).

To tune this balance:

• Compaction Strategy Tuning: Configure async compaction via hoodie.compact.inline=false and run a dedicated compaction engine (e.g., Spark/Flink job). Set trigger policies based on log volume rather than just commit counts using hoodie.compaction.strategy: set BoundedIOCompactionStrategy to cap total compaction work per cycle, preventing write stalls on downstream cluster resources.
• Log File Sizing: Set hoodie.logfile.max.size (e.g., 256MB–512MB) to prevent oversized Avro logs, which slow down memory-intensive inline log merging during Snapshot Queries.
• Payload / Merger Optimization: Implement a record-level HoodieRecordMerger that leverages engine-native vectorization or direct binary merging rather than full Java object deserialization (HoodieRecordPayload).
• Read Optimization via Indexes: Enable hoodie.metadata.index.column.stats.enable to allow snapshot query engines to skip uncompacted delta blocks during query time based on column min/max stats.

Q4: How does Hudi prevent the "Small File Problem" during continuous streaming ingestion?

Answer: Hudi addresses small file generation dynamically at write time using Auto-Bucket Sizing and Bin-Packing.

Incoming Stream Batch
         |
         v
Identify Target Partitions
         |
         v
Find "Small" Files (< 120MB)
         |
         +--------------------------+
         v                          v
Route Inserts to Small Files   Route Updates to Target Files

• Upsert Partitioning (WorkloadProfile): When a batch arrives, Hudi constructs a workload profile categorizing records into Updates (which must go to specific existing File IDs) and Inserts (which can go anywhere).
• Small File Detection: Hudi scans target partitions for existing base files whose sizes are below hoodie.parquet.small.file.limit (default: 100MB, assuming target size is 120MB).
• Bin Packing: Instead of allocating new File Groups for incoming inserts, Hudi routes incoming inserts to be appended/written into those identified small base files until they reach hoodie.parquet.max.file.size.
• Clustering (Asynchronous Reorganization): For MOR tables where inserts spill into tiny delta logs, asynchronous clustering jobs sweep the table out-of-band to merge, re-sort (e.g., Z-Order / Hilbert curves), and rewrite small file groups into uniform 128MB–512MB columnar files without interrupting concurrent writers.

Q5: How are hard deletes processed incrementally in Hudi, and how do you ensure downstream incremental consumers do not miss tombstone events?

Answer:

• Delete Representation: In Hudi, a hard delete is written as a tombstone record with an empty payload or a special delete marker (_hoodie_is_deleted = true). On COW tables, the rewrite omits the deleted key from the new Parquet base file slice. On MOR tables, a HoodieDeleteBlock containing the deleted keys is appended to the active Avro delta log.
• Downstream Incremental Propagation: When an incremental query is executed (query.type = 'incremental'), Hudi scans timeline instants between beginInstantTime and endInstantTime. For MOR tables, the reader processes both data blocks and HoodieDeleteBlock instances, passing the key with _hoodie_is_deleted = true to the downstream engine (e.g., Spark/Flink) so the consuming downstream pipeline can apply the delete.

• Cleaner Safeguards: If Hudi's Cleaner process runs too aggressively and purges old file slices/instants, downstream incremental readers that lag behind will encounter a CommitNotFoundException. To prevent missing tombstones, set hoodie.cleaner.policy = KEEP_LATEST_COMMITS and raise hoodie.cleaner.commits.retained higher than the maximum expected downstream consumer lag (e.g., retaining 24–48 hours of commits).