(Incubating)

The Past, Present and Future

Of�Efficient Data Lake Architectures

Speakers

Vinoth Chandar

• PMC Member
• Co-Creator of project @ Uber
• Background in NoSQL, Stream Processing, Database CDC

Balaji Varadarajan

• PMC Member
• Lead Engineer/Most Active Committer
• Background in large scale distributed systems, Linkedin Databus

Views are our own. Don’t represent views of our employers

Agenda

1) The Elevator Pitch

2) Past : How we built Data Lakes

3) Present : How could we build Data Lakes with Hudi

4) Future : How could do better?

5) Project Status

​

Apache Hudi : Overview

HUDI

Dataset

Why Should You Care?

Near real-time data ingestion to Cloud storage/DFS

Batch jobs on Steroids

Stream processing on batch data

Unified, Optimized analytical storage

GDPR, Data deletions, Compliance.

Building block for great data lakes!

Speaking Technically..

upsert() support with fast, pluggable indexing

Atomically publish data with rollback support

Snapshot isolation between writer & queries

Savepoints for data recovery

Manages file sizes, layout using statistics

Async compaction of row & columnar data

Timeline metadata to track lineage

​

​

Zooming in a bit

Hudi

Dataset

upsert(records) at time t

RDD[Records] : partial inserts/updates

All records updated/created in time range

incrementalPull(t-1, t))

snapshot_query()

at time t

Last committed value for each record in dataset

In the flesh ...

Query

Read Optimized View at 10:10

Query

Real time View at 10:10

Query

Incremental View (10:08, 10:10)

Past : How we built data lakes

Must be Simple?

Database

Events

Apps/

Services

Queries

DFS/Cloud Storage

Extract-Transform-Load

Real-time/OLTP

Analytics/OLAP

External Sources

Tables

But, what about?

• Scaling ingestion to 1000s of tables
• Avoid overwhelming your databases with scans
• Being able to quickly change the ETL
• Measure quality of data in sources
• Data model/field standardization.
• Unlocking all raw data for ML & Data Science

Typical Data Lake

Database

Events

Apps/

Services

Queries

DFS/Cloud Storage

Ingestion

(Extract-Load)

Real-time/OLTP

Analytics/OLAP

External Sources

Raw Tables

Data Lake

Derived Tables

Schemas

Data Audit

An Uber Trip down memory lane

Database

Apps/

Services

Queries

DFS/Cloud Storage

Ingestion

(Extract-Load)

Real-time/OLTP

Analytics/OLAP

Raw Tables

Data Lake

Derived Tables

​

Schemas

Data Audit

Slow & In-efficient

Batch ingestion is too slow..

Rewrite entire tables/partitions several times a day!

ETLs off raw data have no smarts to recompute

Late arriving data is a nightmare

DFS/Cloud Storage

Raw Tables

Data Lake

Derived Tables

120TB HBase table ingested every 8 hrs; Actual change < 500GB

Full recompute every 6-8 hrs

Updated/

Created rows from databases

Streaming Data

Big Big Batch Jobs...

Plus, other unsolved problems

Solving the small file problem, while keeping data fresh(?!)

How to rollback a bad batch of ingestion?

Queries can see dirty data

What if bad data gets through? How to restore dataset?

How to avoid duplicate records in dataset?

Between A Rock & A Hard Place

Lots of equally bad or worse alternatives

1. Query HBase directly
2. Ignore updates to data
3. Only support logs
4. Don’t build a data lake
5. Data modelling tricks

​

​

DFS/Cloud Storage

Raw Tables

Data Lake

Derived Tables

1 raw trip table

100s of derived tables

Hudi Data Lake

Design principles :

• Let’s run mini-batch jobs, in streaming fashion. Move away from big batches.
• Think of it as a database problem
• Pay consideration to 10-100x more data scale & analytical workloads
• Provide different knobs for different trade-offs

​

DFS/Cloud Storage

Raw Tables

Data Lake

Derived Tables

Hudi

upsert()

Hudi

incrementalPull()

Queries

Updated/

Created rows from databases

Present : How to build a Data Lake using Hudi ?

Requirements

1. Incremental Database Ingestion
2. De-duping Log Events
3. Storage Management
4. Transactional Writes
5. Faster Derived/ETL Data
6. Scaling DFS/Storage RPCs
7. Compliance/Data Deletions

Plus much more

• Unique key constraints
• Late data handling
• ...

​

#Q1: Incremental Database Ingestion?

High-value data

• User information in RDBMS
• Business transaction datasets in RDBMS/NoSQL stores

​

Bulk loads don’t scale

• Adds more load to database
• Wasteful re-writing of data

​

MySQL

users

users

Inserts, updates, deletes

Replicate

Data Lake

#A1: Obtain changelogs & upsert()

// Command to extract incrementals using sqoop

bin/sqoop import \

-Dmapreduce.job.user.classpath.first=true \

--connect jdbc:mysql://localhost/users \

--username root \

--password ******* \

--table users \

--as-avrodatafile \

--target-dir \

s3:///tmp/sqoop/import-1/users

​

​

// Spark Datasource

Import org.apache.hudi.DataSourceWriteOptions._

// Use Spark datasource to read avro

Dataset<Row> inputDataset spark.read.avro(‘s3://tmp/sqoop/import-1/users/*’);

​

// save it as a Hudi dataset

inputDataset.write.format(“org.apache.hudi”)

.option(HoodieWriteConfig.TABLE_NAME, “hoodie.users”)

.option(RECORDKEY_FIELD_OPT_KEY(), "userID")

.option(PARTITIONPATH_FIELD_OPT_KEY(),"country")

.option(PRECOMBINE_FIELD_OPT_KEY(), "last_mod")

.option(OPERATION_OPT_KEY(), UPSERT_OPERATION_OPT_VAL())

.mode(SaveMode.Append);

.save(“/path/on/dfs”)

​

Step 1: Extract new changes to users table in MySQL, as avro data files on DFS

(or)

Use data integration tool of choice to feed db changelogs to Kafka/event queue

Step 2: Use your fav datasource to read extracted data and directly “upsert” the users table on DFS/Hive

(or)�Use the Hudi DeltaStreamer tool

#Q2: De-Duping Log Events?

High-scale time series data

• Several billions/day
• Few millions/sec

​

Cause of duplicates

• Client retries/failures/network errors
• At least-once data pipes

​

Overcounting problems

• More impressions => more $
• Low fidelity data

Impressions

Impressions

Produce impression events

Replicate w/o duplicates

Data Lake

#A2: De-Dupe on-the-fly using DeltaStreamer

// Deltastreamer command to ingest kafka events, dedupe, ingest

spark-submit --class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer \

/path/to/hudi-utilities-bundle-*.jar` \

--props s3://path/to/kafka-source.properties \

--schemaprovider-class org.apache.hudi.utilities.schema.SchemaRegistryProvider \

--source-class org.apache.hudi.utilities.sources.AvroKafkaSource \

--source-ordering-field time \

--target-base-path s3:///hudi-deltastreamer/impressions --target-table uber.impressions \

--op BULK_INSERT

--filter-dupes

​

// kafka-source-properties

include=base.properties

# Key fields, for kafka example

hoodie.datasource.write.recordkey.field=event_id

hoodie.datasource.write.partitionpath.field=datestr

# schema provider configs

hoodie.deltastreamer.schemaprovider.registry.url=http://localhost:8081/subjects/impressions-value/versions/latest

# Kafka Source

hoodie.deltastreamer.source.kafka.topic=impressions

#Kafka props

metadata.broker.list=localhost:9092

auto.offset.reset=smallest

schema.registry.url=http://localhost:8081

​

​

​

​

#Q3: Storage Management?

Small Files = Big Problem

• Slow queries
• Stress filesystem metadata

Big Files = Big Delays

• 2GB Parquet writing => ~5-10 mins

File Stitching?

• Non-Standardized
• Consistency?
• Leaks small files to queries anyway!
• High Write amplification!

​

#A3: Deliberate file sizing during write..

Enforce file sizes on write

• No queries on small files!
• Set hoodie.parquet.max.file.size & hoodie.parquet.small.file.limit

​

If you really want to ingest quickly,

• Append to log or small log files
• Transactionally compact into large ones

​

Layout data based on arrival order

• Helps query engines skip/prune files using time ranges.

​

#Q4: Transactional Writes?

Atomic publish of data

• Ingestion can fail midway
• All or nothing

Consistency

• Only Valid data is saved
• Rollback invalid data

Snapshot Isolation

• Read Committed Data
• Concurrent writer/readers

Strong Durability

• No data loss

A

C

I

D

#A4: Multi-row transactions on single dataset

Atomic multi-row commits

• Special .hoodie folder
• Monotonically Increasing timestamps to atomically publish new file versions

Only Valid Data exposed to Queries

• Transparent Rollback of failed writes

Snapshot Isolation

• Using MVCC
• Concurrent Readers, Writer, Compactors

Commit protocol + DFS storage guarantees durability

​

#Q5: Faster Derived/ETL Data?

Multi stage ETL DAGS

• Very common in batch analytics
• Large amount of data

​

​

Derived/ETL tables

• Keep afresh with new/changed raw data
• Star schema/warehousing

​

​

Scaling challenges

• Intelligent recomputations
• Window based joins

​

​

raw_payments

std_payments

standardize_payments(row)

Raw Table

Derived Table

#A5: Streaming Style/Incremental pipelines!

// Spark Datasource

Import org.apache.hudi.{DataSourceWriteOptions, DataSourceReadOptions}._

​

// Use Spark datasource to read avro

Dataset<Row> hoodieIncViewDF = spark.read().format("org.apache.hudi")

.option(VIEW_TYPE_OPT_KEY(), VIEW_TYPE_INCREMENTAL_OPT_VAL())

.option(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY(),

commitInstantFor8AM)

.load(“s3://tables/transactions”);

​

Dataset<Row> stdDF = standardize_payments(hoodieIncViewDF)

​

// save it as a Hudi dataset

inputDataset.write.format(“org.apache.hudi”)

.option(HoodieWriteConfig.TABLE_NAME, “hoodie.std_payments”)

.option(RECORDKEY_FIELD_OPT_KEY(), "id")

.option(PARTITIONPATH_FIELD_OPT_KEY(),"datestr")

.option(PRECOMBINE_FIELD_OPT_KEY(), "time")

.option(OPERATION_OPT_KEY(), UPSERT_OPERATION_OPT_VAL())

.mode(SaveMode.Append);

.save(“/path/on/dfs”)

​

Bring Streaming APIs on Data Lake

​

Incrementally pull

• Avoid recomputes!
• Order of magnitudes faster

​

​

Transform + upsert

• Avoid rewriting all data

​

​

​

​

#Q6: Scaling DFS/Storage RPCs?

Ingestion/Query all list DFS

• List folders/files, take action
• Single threaded vs parallel

​

​

Subtle gotchas/differences

• Cloud storage => no append()
• S3 => Eventual consistency
• S3 => rename() = copy()
• Large directory listings
• HDFS NameNode bottlenecks

#A6: Optimized Timeline/FileSystem APIs

​

Embedded Timeline Server

• 0-listings from Spark executors
• Incremental file-system views on Spark driver

​

Consistency Guards

• Masks eventual consistency on S3
• No data file renames, in-place writing
• Storage aware “append” usage
• Graceful MVCC design to handle various failures

​

​

#Q7: Compliance/ Data Deletions?

Strict rules on data retention

• Delete records
• Correct data
• Raw + Derived tables

​

Need efficient delete()

• Indexed on write (point-ish lookup)
• Still optimized for scans
• Propagate deleted records downstream

​

​

#A7: Efficient/Fast Deletes

Soft deletes

• upsert(k, null)
• Propagates seamlessly via incr-pull

​

​

Hard deletes

• Using EmptyHoodieRecordPayload

​

​

Indexing

• 7-10x faster than using regular joins

Future : How can we do better?

Idea #1 : Smarter Storage Layouts

​

​

​

​

​

​

​

​

Cluster/group records based on query patterns

• Fancier : Ingest in multiple layouts

Use-cases:

• Faster queries due to lesser I/O
• Flexible file-size management

HUDI-112

• Piggy-back on compaction
• MVCC based re-clustering of files
• Allows file sizes for older data to be increased

Idea #2 : Unlock real-time data

(Contd)

Near Real-time data lakes with 1-2 mins data freshness

​

Use-cases:

• Good companion for streaming/real-time datastores
• Act on business data anomalies
• Fresher BI/Dashboards on the data lake

Jira Tickets (log indexing, inserts to log, index performance)

​

• Tune Hudi for near real-time (<1 mins) data ingestion
• Index Lookup Speedup : Guidance for record-key selections to leverage range-pruning
• Improve Ease of deployment and tooling for compaction management.

Idea #3 : Universal file format/runtime support

Standardization should cover all major tools/environments

Use-cases :

• Lots of data across avro/parquet/csv/orc
• Lots of processing frameworks : Spark, Flink, Beam, ...

​

(ORC PR/JIRA, Flink discussions,)

• Ongoing work to support ORC file format (HIP #2, PR #657)
• Wide interest in Community to support Flink as processing engine (Discussion Threads)

Idea #4 : Scalable Metadata Management

HUDI-241 (bundle, and new JIRA for standalone timeline server)

• Provide fs listings for both query planning and ingestion
• Idea: Track column level statistics in each file and help query planner

​

Leveraging granular metadata about dataset to plan queries and cut down DFS rpcs

Use-cases :

• Move beyond just partition level metadata in Hive metastore -> file-level metadata
• Potentially evaluate pushdowns more efficiently e.g: avoid reading all footers
• Amortize the cost of large directory listings by caching metadata

Idea #5 : Bootstrap historical data

Okay, all these ideas are great! How do I get it on my existing data?

Use-cases

• Easy migration, without rewriting a ton of data
• Allow for safe experimentation

HUDI-242 (migration story)

• Already supports partition level migration
• Also support indexing of older partitions
• Older partitions “untouched” and “unmoved”
• Seamless tooling way to “bootstrap” and continue ingesting

​

Hudi : State-Of-The-Union

Hudi : Open Sourcing & Evolution..

2015 : Published core ideas/principles for incremental processing (O’reilly article)

2016 : Project created at Uber & powers all database/business critical feeds @ Uber

2017 : Project open sourced by Uber & work begun on Merge-On-Read, Cloud support

2018 : Picked up adopters, hardening, async compaction..

2019 : Incubated into ASF and working towards first release

Community

​

Diverse PMC members with experience building data systems at scale

Mentors from Amazon/Lyft

Adopted by few companies, 5x more in testing/evaluation

Fast growing, with a challenging & interesting technical roadmap

Seeking solid, diverse new contributors

Dev mailing lists

Spark bundle downloads

Github Chatter

Project Management

Open collaboration in the “Apache” way

HIP process for design reviews of large features

Bug reports/Support via Github Issues

JIRAs for project issue management

Github PRs for code reviews & contributions

Resources

User Docs : https://hudi.apache.org

Technical Wiki : https://cwiki.apache.org/confluence/display/HUDI

Github : https://github.com/apache/incubator-hudi/

Twitter : https://twitter.com/apachehudi

Mailing list(s) : dev-subscribe@hudi.apache.org (send an empty email to subscribe)

dev@hudi.apache.org (actual mailing list)

Slack : https://join.slack.com/t/apache-hudi/signup

​

​

​

Big Picture

Fills a clear void in data ingestion, storage and processing!

Leads the convergence towards streaming style processing!

Brings transactional semantics to managing data

Positioned to solve impending demand for scale & speed

Evolve as “de facto”, open, vendor neutral standard for data storage!

Thanks!

Questions?