Uber: from a day-old data lake to minutes

At a glance

Organization	Uber
Period	About 2014 to 2018
Problem	Analytical data was a day old, and the lake couldn’t handle updates
Architecture	A Hadoop (opens in a new tab) data lake with Hudi (opens in a new tab) tables that support upserts and incremental reads, fed from Kafka (opens in a new tab)
Patterns	Data warehouse, data lake, lakehouse, change data capture

How the platform evolved

Generation 1: a data warehouse. Uber’s data lived in a few operational MySQL (opens in a new tab) and PostgreSQL (opens in a new tab) databases. To get one view of the business, Uber built a warehouse on Vertica, loaded by ad hoc ETL jobs. It worked for SQL users but had no schema contract for incoming JSON, ingestion was fragile, and it was expensive to scale, so data was deleted to save space.

Generation 2: a Hadoop data lake. Raw data was ingested once, untransformed, into Hadoop and read with Presto, Spark (opens in a new tab) and Hive (opens in a new tab), stored as Parquet (opens in a new tab). It grew to tens of petabytes and more than 100,000 batch jobs a day. But:

• Data was about 24 hours old.
• Each ingestion re-read whole snapshots, taking over 20 hours on more than 1,000 Spark executors.
• HDFS and Parquet couldn’t update data in place, yet much of Uber’s data, such as trips, changes after it is created.
• Huge numbers of small files strained the cluster.

Generation 3: incremental processing with Hudi. Uber built Hudi, an open-source library on Spark that adds two abilities to files in the lake:

• Upserts and deletes: insert, update or delete records in existing Parquet data.
• Incremental reads: a consumer asks for everything changed since a checkpoint, instead of scanning whole tables.

An ingestion framework, Marmaray, reads change logs from Kafka every 10 to 15 minutes and applies them to Hudi tables. Each dataset gets a change-history table and a merged, latest-state table.

flowchart LR
    db[(Operational databases)] -- changes --> kafka{{Kafka}}
    kafka --> ing[Marmaray: every 10 to 15 min]
    ing --> hist[(Change-history table)]
    ing --> latest[(Latest-state table, Hudi upserts)]
    latest -- incremental reads --> etl[Downstream ETL]
    latest --> sql[Presto, Spark, Hive]

Results

Uber reported more than 100 petabytes in the platform and end-to-end latency for derived tables cut from 24 hours to under one hour, with raw data about 30 minutes old. Its next targets, as of 2018, were five-minute latency for raw data and ten minutes for modeled tables.

Trade-offs

• More moving parts than plain files: commit timelines, compaction and table services must be run.
• Write cost. Updating records in columnar files means rewriting files or merging changes at read time.
• Tooling maturity at the time. Uber had to build Hudi and Marmaray itself; today several open table formats offer these features.

Lessons you can reuse

• If your data changes after creation, your lake needs upserts. Append-only files force full reloads.
• Process incrementally: read only what changed since the last run.
• Feed the lake from change logs in Kafka rather than repeated snapshots of source databases.

Related on ArchitectureCompass

• Lakehouse, whose reference implementation applies the same ideas with Apache Iceberg (opens in a new tab)
• Data lake vs lakehouse
• LinkedIn builds Kafka

Sources

• Reza Shiftehfar, “Uber’s Big Data Platform: 100+ Petabytes with Minute Latency” (opens in a new tab), Uber Engineering, October 2018