Data Lakes (1) - How Hudi's Core Features Work

As internet businesses mature and warehouses and model training become established, more engineers shift from feature development to architectural upgrades. Hudi and Iceberg frequently emerge as alternatives to Hive/HDFS-based architectures.

• Data Lakes (1) - How Hudi’s Core Features Work
• Data Lakes (2) - How Iceberg’s Core Features Work

Overview

Many online comparisons emphasize Iceberg’s schema support and Hudi’s upserts. Much of this is already outdated: key features will likely converge over the coming months. Understanding their backgrounds, design ideas, and implementation details is therefore valuable.

This article surveys Hudi’s mechanisms without going into source-code details, giving readers unfamiliar with data lakes an overview of the technology and its strengths.

Apache Hudi

Hudi, Hadoop Upsert anD Incremental, originated at Uber. It initially addressed data consistency in Lambda warehouse architectures, replacing streaming with incremental processing and introducing two important features: upsert and incremental pull.

A flagship use case matches rider requests with driver acceptances. Hudi’s upsert and incremental-read capabilities join these two streams at minute-level latency to produce rider-driver matches.

Two years ago, Hudi called itself a pipeline or storage framework. Its website now describes a next-generation streaming data lake platform. Commercialization has expanded its ambitions considerably.

Basic Concepts

Timeline

A timeline records a Hudi table’s information and actions at different times. Managed by TimelineServer and backed by persistent storage such as HDFS or an RDBMS, its information actually resides in the table’s .hoodie directory, with filenames distinguishing instants. It enables version management and incremental, time-based processing.

Three concepts define the timeline:

• action: An operation such as commit or rollback.
• time: Its timestamp, at millisecond resolution.
• state: The action’s status.

Every metadata-changing operation submits an action to the timeline. These operations must be atomic and generally coordinated at one point. With Spark and Flink, their Driver and JobMaster respectively record timeline information.

Table Types & Query Types

Hudi offers Copy-on-Write and Merge-on-Read tables, with these query types:

Copy-on-Write updates rewrite the containing file. Write amplification is high, but read amplification is zero, suiting read-heavy workloads. Two queries are supported:

• Snapshot Query: Reads the latest snapshot and therefore the latest data.
• Incremental Query: Scans records and retains those with commit_time greater than the supplied commit time.

The following GIF illustrates the flow:

Merge-on-Read resembles an LSM-tree. Writes enter row-oriented delta data, which can be manually merged into existing columnar Parquet files. Three queries are supported:

• Snapshot Query: Reads the latest snapshot, combining row-oriented and columnar data.
• Incremental Query: Filters records by commit_time greater than the supplied time, combining row-oriented and columnar data.
• Read Optimized Query: Reads base files only, excluding deltas. Columnar storage makes this efficient.

The following GIF illustrates the flow:

MOR tables contain base files and log files. Base files are usually columnar Parquet, optimized for reads; log files usually use row-oriented Avro, optimized for writes.

Index

Since Hudi tables have primary keys, indexes naturally locate data for more efficient reads and writes. Different index types provide different granularities:

• Bloom Index
• Simple Index
• HBase Index
• Hash Index

For each record, querying or calculating its primary-key index determines whether it is an insert or update and locates the existing file. Index lookup is crucial for streaming writes: its efficiency directly affects throughput and stability. This topic could merit a separate article.

File Layouts

From outside to inside, the layout is:

• Table
• Partition
• FileGroup, identified by FileGroupId or FileID: Each partition contains multiple file groups, each with a base file and several log files.
• Base and log files: As described for MOR tables above.

Core Mechanisms

Upsert

Upserts depend on index type. Interestingly, Hudi’s original Spark-centric architecture did not consider other engines, producing major differences in how connectors use indexes.

Using the supported Spark and Flink engines as examples, here is how upsert works:

• Spark：
  1. Deduplicate records by primary key. If Payload implements preCombine, merge matching payloads with it; otherwise retain the first matching record encountered.
  2. Call the index’s tagLocation to find existing records by primary key, recording fileId and commitTs. Missing records temporarily receive a null location.
  3. Count records by partition and create WorkloadStat with insert/update counts. Use existing file distribution, preferring small file groups for new data, to determine each file group’s workload.
  4. Assign file groups: updates use indexed locations, while inserts are distributed according to the previous step. partitionBy file-group location so each Spark partition processes one file group.
  5. Write each Spark partition, return successful record locations, update the index record by record, and commit the write to Hudi’s timeline after index updates succeed.
• Flink：
  1. A new checkpoint starts a corresponding Hudi instant. The interval between successful checkpoints represents one instant’s writes.
  2. keyBy primary key so matching records reach the same task.
  3. Store each key’s location in Flink state. Updates read the existing file group from state; inserts choose a location from the file distribution, similarly to Spark.
  4. keyBy FileGroup ID so records for one location reach the same task.
  5. Writer tasks buffer and write batches, then send successful-write metadata to JobMaster for commit at checkpoint time.

Bulk Insert

Bulk insert initializes a partition or table. With no updates to consider, it is much faster than upsert.

Incremental Reads

See Hudi’s official incremental-query example.

1
// spark-shell
2
// reload data
3
spark.
4
  read.
5
  format("hudi").
6
  load(basePath).
7
  createOrReplaceTempView("hudi_trips_snapshot")
8

9
val commits = spark.sql("select distinct(_hoodie_commit_time) as commitTime from  hudi_trips_snapshot order by commitTime").map(k => k.getString(0)).take(50)
10
val beginTime = commits(commits.length - 2) // commit time we are interested in
11

12
// incrementally query data
13
val tripsIncrementalDF = spark.read.format("hudi").
14
  option(QUERY_TYPE_OPT_KEY, QUERY_TYPE_INCREMENTAL_OPT_VAL).
15
  option(BEGIN_INSTANTTIME_OPT_KEY, beginTime).
16
  load(basePath)
17
tripsIncrementalDF.createOrReplaceTempView("hudi_trips_incremental")
18

19
spark.sql("select `_hoodie_commit_time`, fare, begin_lon, begin_lat, ts from  hudi_trips_incremental where fare > 20.0").show()

Each table has a hidden _hoodie_commit_time timestamp column. Users define an incremental range by commit timestamps. Internally, beginTime filters timeline instants, whose write metadata identifies files to scan. If compaction has occurred, records in compacted files are filtered by the requested time range.

Other Features

• Transactions/concurrency: Initially, isolation was table-wide, so only one of two concurrent commits to a table succeeded. Later improvements reduced conflict granularity to file groups, a common database technique.
• preCombine: Each record is wrapped in HoodieRecordPayload. Users can define merging for records sharing a primary key; the default keeps the latest record.

1
public interface HoodieRecordPayload<T extends HoodieRecordPayload> extends Serializable {
2
  T preCombine(T oldValue);
3
}

Summary

This article covered the timeline, COW/MOR tables, file organization, upserts, and incremental reads. Hudi acts as a feature-rich format, using merge-on-read ideas to provide upsert semantics.

The next article will cover the Iceberg stack.