A few introductory notes on Apache Hudi.

Background

Hudi is an open-source data lake framework led by Uber. Many design choices therefore come from Uber’s own use cases, such as joining driver and rider data by order ID. Like many companies, Uber previously used a Lambda architecture combining batch and streaming. Let us compare the two models in terms of latency, data completeness, and cost.

Batch Model

Batch processing uses engines such as MapReduce, Hive, or Spark to run hourly or daily jobs.

• Latency: Hours or days. This includes the scheduling interval, execution of dependent jobs, and the time until results appear in the data platform. With large datasets and complex logic, an hourly job often has an actual latency of two to three hours.
• Data completeness: Relatively high. For processing-time calculations, an hourly job generally receives all data from that hour. Event-time requirements are harder because client-side reporting can be delayed, limiting the usefulness of batch jobs.
• Cost: Low, because resources are occupied only while jobs run and can otherwise be lent to online services. From another perspective, cost can be high: changes to source data or late arrivals require a full recomputation.

Streaming Model

Flink is a typical engine for real-time stream processing.

• Latency: Very low, potentially real time.
• Data completeness: Lower. Streaming engines begin processing before all data arrives. Watermarks establish a cutoff, and data older than the watermark is discarded, so completeness cannot be absolutely guaranteed. Internet businesses often use streaming for live campaign dashboards, where completeness requirements are modest. In many cases, users maintain two programs: a streaming job for immediate results and a batch job to correct them the next day.
• Cost: High. Streaming jobs run continuously, and joins across streams generally require state in memory or a database. Serialization overhead and additional I/O with external components become significant at scale.

Incremental Model

Uber proposed an incremental model to balance these tradeoffs: more timely than batch and more economical than streaming.

In simple terms, the incremental model runs near-real-time jobs as mini-batches. Hudi supports two key features:

• Upsert: Addresses the inability to insert and update data in the batch model. Incremental changes can be written to Hive instead of overwriting everything. Hudi maintains a key-to-file mapping, making the relevant file easy to locate for an upsert.
• Incremental Query: Reduces the amount of input processed. For Uber’s driver-rider stream join, each run can fetch the new data from both streams and perform a batch join, reducing cost by orders of magnitude compared with streaming.

Hudi provides two table types for this model: Copy-On-Write and Merge-On-Read.

Copy-On-Write Table

In a Copy-On-Write table, updates rewrite the file containing the data. Write amplification is high, but read amplification is zero, making it suitable for read-heavy workloads. It supports two query types:

• Snapshot Query: Reads the latest snapshot, representing the latest data.
• Incremental Query: The user supplies a commit time. Hudi scans records and retains those whose commit_time is greater than that time.

The following GIF illustrates the process:

Merge-On-Read Table

A Merge-On-Read table resembles an LSM-tree. Writes first enter row-oriented delta data, which can be manually merged into existing files stored in columnar Parquet format. It supports three query types:

• Snapshot Query: Reads the latest snapshot, combining row-oriented and columnar data.
• Incremental Query: Scans records and filters for commit_time greater than the user-specified time, combining row-oriented and columnar data.
• Read Optimized Query: Reads only existing base data, excluding deltas. Columnar files make this efficient.

The following GIF illustrates the process:

Thoughts

Hudi provides a convenient Docker demo for quickly trying these features.

Discussions of data lake frameworks usually mention Delta Lake, Apache Hudi, and Apache Iceberg. Although frequently compared, they come from different backgrounds.

Iceberg originally addressed inconsistent file formats at Netflix. Hive tables might contain CSV or Parquet, and metadata changes required users to understand numerous table properties. Iceberg proposed that everything could be a table, aiming to unify them under Iceberg tables.

Hudi instead offers a compromise between batch and streaming. I know less about Delta, but its overall approach resembles Hudi’s. Apache Iceberg is also actively developing row-level updates, similar to Hudi’s upsert capability.

Despite their different starting points, all three challenge Hive, which has dominated data warehousing for years while changing relatively little. The Hadoop ecosystem now supports much greater data volumes and variety. Hive makes a warehouse easy to build, but has limited control over its contents. As an analogy, Hive supplies a warehouse without using knowledge of the goods to optimize its operation. People put things inside, but finding them becomes chaotic as it fills. New data lake frameworks add labels, monitoring, and intelligent transport, allowing users to locate the right shelf even in a crowded warehouse.

References

• Youtube Video: Apache Hudi
• Hudi Design