Adoption

• In use of Yahoo!, Foursquare, Berkeley, Princetion & many others(possibly Taobao, Netease)
• 600+ member meetup, 800+ watcherson Github
• 30+ contributors

How do you scale up applications to PBs of data?

I can use hundreds or thousands of machines!

But distributed programming is hard(task scheduling, data synchronization, machines failures)

Dterminism & Idempotence

Map and reduce functions are:

• deterministic
• side-effect free

Tasks are thus idempotent:

• Rerunning them gets you the same result

Dterminism & Idempotence

Fault-tolerance:

• Rerun tasks originally schedualed on failed nodes

Straggers:

• Rerun tasks originally schedualed on slow nodes

Why go Beyond MapReduce?

MapReduce simplified big data analysis by giving a reliable programming model for large clusters

But as soon as it got popular, users wanted:

• More complex, multi-stage applications
• More interactive, ad-hoc queries

Complex jobs and interactive queries both need one thing that MapReduce lacks:

Efficient primitives for data sharing

In MapReduce, the only way to share data across jobs is stable storage(e.g. HDFS)

SLOW!

Programming Model

Resilient distributed datasets(RDDs)

• Immutable, partitioned collections of objects
• Can be cached in memory for effcient reuse

Transformations(e.g. map, filter, groupBy, join)

• Build RDDs from other RDDs

Actions(e.g. count, collect, save)

• Return a result or write it to storage

Fault Tolerance

RDDs track the series of transformations used to build them (their lineage) to recompute lost data

Fault Recovery Results

Tradeoff Space

Behavior with Not Enough RAM

Logistic Regression Performance

Implementation

Use Mesos/YARN to share resources with Hadoop & other frameworks

Can access any Hadoop input source (HDFS, S3, …)

20k lines of code

User Applications

• In-memory analytics & anomaly detection (Conviva)
• Interactive queries on data streams (Quantifind)
• Exploratory log analysis (Foursquare)
• Traffic estimation w/ GPS data (Mobile Millennium)
• Twitter spam classification (Monarch)

Conviva GeoReport

Group aggregations on many keys w/ same filter

40× gain over Hive from avoiding repeated reading, deserialization and filtering

Challenges

• Data volumes expanding
• Faults and stragglers complicate parallel database design
• Low-latency, interactivity

MPP Databases

• Vertica, SAP HANA, Teradata, Google Dremel...
• Fast!
• Generally not fault-tolerant; challenging for long running queries as clusters scale up.
• Lack rich analytics such as ML and Graph algorithms.

MapReduce

• Apache Hive, Google Tenzing, Turn Cheetah...
• Deterministic, idempotent tasks: enables fine-grained fault-tolerance and resouce sharing.
• Expressive ML algorithms.
• High-latency, dismissed for interactive workloads.

Shark

A data warehouse that

• builds on Spark,
• scals out and is fault-tolerance,
• supports low-latency, interactive queries through in-memory computation,
• supports both SQL and complex analytics,
• is compatible with Apache Hive(storage, serdes, UDFs, types, metadata).

Hive Architecture

Shark Architecture

Engine Features

• Columnar Memory Store
• ML Integration
• Partial DAG Execution
• Data Co-partitioning
• Partition Pruning based on Range Statistics

Efficient In-Memory Storage

Simply caching Hive records as Java objects is inefficient due to high per-object overhead

Instead, Shark employs column-oriented storage using arrays of primitive types.

Benefit: similarly compact size to serialized data, but>5x faster to access

ML Integration

• Unified system for query processing and machine learning
• Query processing and ML share the same set of workers and caches

Performance

Task Launch Overhead

Hadoop uses heartbeat to communicate scheduling decisions.

Task launch delay 5-10 seconds.

Spark uses an event-driven architecture and can launch tasks in 5ms

• better parallelism
• easier straggler mitigration
• multi-tenancy resouce sharing

Task Launch Overhead