Apache Hudi™ at Uber: Engineering for Trillion-Record-Scale Data Lake Operations

Prashant Wason, Balajee Nagasubramaniam, Surya Prasanna Kumar Yalla, Meenal Binwade, Xinli Shang, Jack Song
19 min readadvanced
--
View Original

Overview

This article details how Uber built and scaled Apache Hudi to power one of the world's largest data lakes, managing 19,500 datasets with trillions of records across a multi-hundred-petabyte repository. It covers the engineering motivations behind creating Hudi, key innovations like the Metadata Table and Record Index that enabled trillion-record-scale operations, and lessons learned from nearly a decade of operating extreme-scale data lake infrastructure.

What You'll Learn

1

Why traditional append-only, batch-oriented data lake architectures fail at Uber-scale mutable workloads and what primitives are needed to fix them

2

How to design a Metadata Table (MDT) with HFile-backed key-value storage to eliminate file listing bottlenecks at extreme scale

3

How to implement a Record Index for O(1) key lookups across trillion-record tables, replacing external dependencies like HBase

4

How to architect multi-data-center replication for data lake tables with consistent failover and cloud migration support

5

Why configuration management, schema evolution, and automated observability are critical operational concerns at scale

Prerequisites & Requirements

  • Understanding of data lake concepts including ACID transactions, partitioning, and file formats like Parquet
  • Familiarity with distributed processing frameworks such as Apache Spark and Apache Flink
  • Understanding of upsert operations, indexing strategies, and incremental processing patterns
  • Experience with large-scale data engineering or data platform operations(optional)

Key Questions Answered

Why did Uber build Apache Hudi instead of using existing data lake technologies?
By 2015, Uber's data systems were expanding faster than off-the-shelf technologies could handle. The industry-standard data lake was append-only and batch-oriented, which couldn't support Uber's mutable data models, high-volume update streams, or end-to-end incremental processing needs. Uber needed database-like primitives—ACID transactions, fast upserts with indexing, and incremental processing—while retaining data lake scalability and flexibility.
How does the Hudi Metadata Table solve the file listing bottleneck at scale?
The Metadata Table (MDT) is an HFile-backed key-value store within Hudi that tracks file listings, column statistics, and bloom filters. It provides O(1) lookups by mapping directory paths to specific keys, eliminating the need to hit the HDFS NameNode for file listings. This removed the bottleneck caused by frequent file listings from writers, readers, table services, and query engines across tens of thousands of datasets with millions of files.
How does the Hudi Record Index enable efficient upserts on trillion-record tables?
The Record Index is an HFile-backed data structure stored within the Metadata Table that maps record keys directly to file groups, enabling O(1) key lookups without external servers like HBase. For tables exceeding 300 billion rows, the index is sharded into up to 10,000 HFiles for parallelized lookups. Lookup latency is 1-2 milliseconds per record key, and initialization for a 300-billion-row table takes approximately 7 hours using 4,000 executors.
What are the different workload types in Uber's data lake and how many tables does each have?
Uber classifies datasets into four workload types: append-only datasets (11,200 tables) for high-volume raw ingestion using bulk inserts; upsert-heavy datasets (4,400 tables) for mutable business states requiring index-based lookups; derived datasets (1,600 tables) for analytics and ML through transformations and aggregations; and realtime ingestion via Flink-native streaming (500 tables) for sub-15-minute freshness SLOs.
How does Uber handle multi-data-center replication for Hudi tables?
Uber uses a primary-secondary dataset design where Hudi's commit timeline and atomic operations propagate every write safely to a secondary region. Validation and consistency-checking tools ensure replicas remain aligned. A table availability service monitors regional health and promotes the secondary dataset during disruptions, while intelligent query routing directs reads to the nearest or healthiest region for improved latency and compute balancing.
What is Uber's IngestionNext initiative and how does Hudi support it?
IngestionNext is Uber's initiative to shift from batch to streaming ingestion. Hudi bridges streaming runtime and storage by embedding Flink checkpoint IDs in commit metadata for deterministic recovery, preventing data duplication during failovers. Hudi's flexible file management enables row-group-level merging that accelerates compaction by over 10x, solving the small-file problem in streaming. This reduced data latency from hours to minutes while cutting compute usage.
What are the key operational lessons from running a trillion-row data lake?
Five critical lessons emerged: treat schema evolution as a first-class concern with automated validation; plan around the hardest technical challenges first rather than deferring them; invest in constant automated observability and validation to catch silent issues early; centralize configuration management to avoid tens of thousands of subtly different configs becoming a bottleneck; and design for fast, safe recovery since failures are inevitable at scale.
What is Uber's future roadmap for Apache Hudi?
Uber is engaging with Onehouse to accelerate Hudi 1.x adoption across three pillars: foundational modernization (porting internal commits to open source and deprecating legacy RDD APIs), performance optimization (clustering efficiency improvements, dynamic MDT sharding, and evaluating dual-writing to Apache Iceberg), and capability expansion (scaling Flink-based ingestion with Record Level Index, enhancing DeltaStreamer for self-service, and adopting modern Presto connectors).

Key Statistics & Figures

Total Hudi datasets managed
19,500
Across multiple business domains at Uber
Rows ingested per day
6 trillion
Daily ingestion volume across Uber's data lake
New data files added per day
3 million
Daily new files added to the data lake
Total logical storage
350 petabytes
On HDFS and Google Cloud Storage
Data ingested per day
10 petabytes
With over 3 petabytes written to the data lake
Commits per day
350,000
Daily Hudi commit operations
Table service operations daily
70,000
Compactions, cleans, and clustering operations
Analytical queries per week
4 million
Served via Presto and Apache Spark
Ultra-large table size
400+ billion rows
Largest tables that update continuously
Record Index lookup latency
1-2 milliseconds
Per single record key from an HFile
Record Index initialization time
~7 hours
For 300 billion row tables using ~4,000 executors
Record Index HFile shards (max)
10,000
For tables exceeding 300 billion rows
MDT deployment coverage
Over 90%
Percentage of production datasets using Metadata Table
Append-only tables
11,200
Majority of Uber's datasets, highest ingestion volume
Upsert-heavy tables
4,400
Model changing business states with high update frequencies
Derived datasets
1,600
Power analytics and ML through transformations
Realtime ingestion tables
500
Flink-native streaming with sub-15-minute freshness
Record Index tables in production
3,600
Large tables requiring frequent record-level modifications
Compaction speedup with IngestionNext
10x
Row-group-level merging strategy acceleration
GitHub stars
6,030
Apache Hudi community
Community members
15,000+
Across Hudi Slack and GitHub
Pull requests
10,000+
Apache Hudi project activity
Contributing companies
50+
Per ossinsight.io
Cloud providers supporting Hudi
10
Including AWS, Azure, and GCP
Fortune 500 companies visiting Hudi website
65%
Out of 36,000+ companies visiting the site
PMC members
20
Apache Hudi project governance
Committers
41
Apache Hudi project
Contributors
2,000+
Apache Hudi project

Technologies & Tools

Some links below are affiliate links. We may earn a commission if you make a purchase.

Data Lake Storage Engine
Apache Hudi
Core data lake table format providing ACID transactions, indexing, upserts, and incremental processing
Distributed Processing
Apache Spark
Batch ingestion, ETL pipelines, feature engineering, distributed batch ML workloads, and query engine
Stream Processing
Apache Flink
Continuous low-latency streaming ingestion with sub-15-minute freshness SLOs
Query Engine
Presto
High-concurrency interactive analytics queries on Hudi tables
Distributed Storage
Hdfs
Primary distributed file storage for the data lake
Cloud Storage
Google Cloud Storage
Cloud storage for data lake with dynamic scaling capabilities
File Format
Apache Parquet
Columnar file format for data lake storage
File Format
Hfile
SSTable-based format used for Metadata Table and Record Index with fast indexed lookups
Database
Hbase
Previously used as external index for record lookups, replaced by native Record Index
Data Lake Table Format
Apache Iceberg
Being evaluated for dual-writing capability via Hudi 1.x pluggable table formats
Monitoring
M3
Metrics system for tracking Hudi operations including read/write performance, commit latency, and compaction throughput

Key Actionable Insights

1
Implement a metadata layer to eliminate filesystem-level bottlenecks for large-scale data lake operations. Storing file listings, column statistics, and bloom filters in an indexed key-value store within the table format itself allows O(1) lookups instead of expensive filesystem list operations. This is especially critical when managing thousands of partitions with millions of files.
Uber deployed their Metadata Table to over 90% of production datasets, fundamentally changing the economics of operating their data lake by removing dependence on HDFS NameNode for file listings.
2
Build native record-level indexing into your data lake storage layer rather than relying on external index services like HBase. An embedded index that maps record keys to file groups enables O(1) lookups without external operational dependencies, reducing both latency and infrastructure complexity for upsert-heavy workloads.
Uber's Record Index achieves 1-2 millisecond lookup latency per record key and supports sharding into up to 10,000 HFiles for tables exceeding 300 billion rows, enabling parallelized lookups across thousands of executors.
3
Tackle the hardest technical challenges in your data platform first rather than deferring them. Building scalable indexing, schema evolution, and mutable data support early prevents architectural dead-ends that become exponentially harder to fix as the platform grows and more dependencies accumulate.
Uber learned that while small deployments can use bloom filters and file scanning for upserts, these approaches become completely intractable at trillion-row scale. Without planning for scalable indexing from the start, no amount of compute would make such workloads viable.
4
Centralize configuration management for your data lake tables to prevent configuration drift from becoming a scaling bottleneck. What starts as a few parameter overrides per dataset quickly turns into tens of thousands of subtly different configurations, making it impossible to roll out improvements safely.
Uber's Hudi Config Store provides a controlled, globally consistent configuration layer that enables new features, compaction logic, and file-sizing improvements to be rolled out across thousands of datasets safely and predictably.
5
Treat schema evolution as a first-class concern from day one by implementing strong backward-compatibility rules, automated schema validation, and pre-change validation. At scale, a seemingly harmless field rename can break thousands of downstream jobs, and the operational cost of schema issues grows exponentially with dataset count.
Uber's schemas change constantly as product teams iterate and launch new features. Without automated guardrails, schema changes became one of the most persistent sources of operational incidents.
6
Invest in automated observability and validation tooling that continuously checks data lake consistency, rather than relying on reactive incident response. Silent issues like corrupted files, configuration errors, and open source bugs are the most dangerous because they can cascade into large-scale outages before detection.
Uber's Hudi Validation tool detects inconsistencies early, and their M3-based metrics system tracks commit latency, compaction throughput, file growth rates, and table health across all 19,500 datasets.

Common Pitfalls

1
Using filesystem-level file listings at scale. As data lakes grow to thousands of partitions and millions of files, direct filesystem calls (e.g., HDFS NameNode listing) become a critical bottleneck, causing significant latency and operational instability for writers, readers, table services, and query engines.
Uber solved this by creating the Metadata Table to store file listings in an indexed key-value store, enabling O(1) lookups instead of filesystem traversals.
2
Relying on bloom filters or file scanning for upserts at large scale. While these approaches work for small deployments, they become completely intractable for trillion-record tables where record-level lookups need to be deterministic and fast across hundreds of partitions and billions of records.
The Record Index was developed to provide O(1) key lookups with 1-2ms latency, replacing approaches that could not scale to Uber's upsert-heavy workloads.
3
Deferring hard technical challenges like indexing and mutable data support in favor of easier problems. This strategy can seriously derail the platform later, as architectural limitations compound and become exponentially more expensive to fix once thousands of datasets and downstream consumers depend on the existing design.
Uber explicitly calls this out as a key lesson: plan the tech stack around the toughest challenges first.
4
Allowing per-dataset configuration overrides to grow unchecked. What starts as a few parameter tweaks quickly turns into tens of thousands of subtly different configurations, making it impossible to safely roll out improvements, debug issues, or maintain consistency across the platform.
Uber's centralized Hudi Config Store provides a globally consistent configuration layer that enables safe, predictable rollouts across thousands of datasets.
5
Treating schema changes as routine operations. At scale, even seemingly harmless changes like field renames can break thousands of downstream jobs. Without strong backward-compatibility rules, automated schema validation, and pre-change validation, the operational cost of schema evolution grows exponentially.
Uber treats schema evolution as a first-class concern with automated guardrails to prevent cascading failures across their data ecosystem.
6
Depending on external services like HBase for critical index lookups, introducing operational complexity and additional failure domains. External dependencies create latency, require separate capacity planning, and add failure modes that compound at scale.
Uber moved from an external HBase index to a native Record Index within Hudi's Metadata Table, eliminating this significant operational dependency.

Related Concepts

Data Lakehouse Architecture
Acid Transactions In Data Lakes
Incremental Data Processing
Upsert Operations At Scale
Copy-on-write Vs Merge-on-read Table Types
Data Lake Table Formats (hudi, Iceberg, Delta Lake)
Stream Processing With Apache Flink
Batch Vs Streaming Ingestion
Multi-data-center Replication
Schema Evolution And Backward Compatibility
Data Freshness Slas
File Compaction And Clustering
Distributed Indexing Strategies
Data Lake Observability And Monitoring
Configuration Management At Scale
Disaster Recovery For Data Platforms