Source-linked AI summary
Shark: SQL and Rich Analytics at Scale
Reynold Xin, Josh Rosen, Matei Zaharia, Michael J. Franklin, Scott Shenker, Ion Stoica
TL;DR
Large-scale analysis must support interactive SQL, sophisticated analytics, and fault recovery as data and clusters grow. Shark combines these capabilities in an RDD-based MapReduce-like engine with SQL optimizations, achieving up to 100× speedups while retaining fine-grained recovery.
Problem
Large-scale analysis needs interactive SQL and sophisticated analytics while handling increasing faults and stragglers across growing clusters.
Method
Shark extends a MapReduce-like RDD engine with fine-grained lineage recovery, column-oriented in-memory storage, and partial DAG execution for runtime query re-optimization.
Results
Shark is up to 100× faster than Hive for SQL and 100× faster than Hadoop for machine learning, while recovering from failures mid-query within seconds.
Takeaways & Limitations
Shark demonstrates that MapReduce-like execution can combine SQL, complex analytics, and fine-grained mid-query fault tolerance in one runtime.
Takeaways & Limitations
Shark is not claimed to be fundamentally faster than MPP databases and has implementation disadvantages including JVM execution.
Abstract
from arXiv · showhide
Shark is a new data analysis system that marries query processing with complex analytics on large clusters. It leverages a novel distributed memory abstraction to provide a unified engine that can run SQL queries and sophisticated analytics functions (e.g., iterative machine learning) at scale, and efficiently recovers from failures mid-query. This allows Shark to run SQL queries up to 100x faster than Apache Hive, and machine learning programs up to 100x faster than Hadoop. Unlike previous systems, Shark shows that it is possible to achieve these speedups while retaining a MapReduce-like execution engine, and the fine-grained fault tolerance properties that such engines provide. It extends such an engine in several ways, including column-oriented in-memory storage and dynamic mid-query replanning, to effectively execute SQL. The result is a system that matches the speedups reported for MPP analytic databases over MapReduce, while offering fault tolerance properties and complex analytics capabilities that they lack.
1 Introduction
Shark addresses the need for interactive analysis at growing scale by combining SQL, complex analytics, and fine-grained fault tolerance in a MapReduce-like engine. Its optimizations deliver large speedups while preserving recovery capabilities.
- Modern analysis must handle expanding data volumes, increasing faults and stragglers, sophisticated methods, and interactive-speed queries.
- Shark combines SQL processing and complex analytics with fine-grained fault recovery across both operation types.
- RDD lineage enables Shark to recompute lost data partitions and recover mid-query without replicating every partition.
- Shark supports Hive SQL and UDFs, Spark-based analytics, shared in-memory execution, and recovery within a single execution plan.
- 100× faster than Hive for SQL and 100× faster than Hadoop for iterative machine learning, with mid-query recovery within seconds.
- MapReduce-like execution can support SQL effectively while combining relational processing with complex analytics and fine-grained fault tolerance.
2 System Overview
Shark builds SQL and analytics processing on Spark’s RDD abstraction, using immutable partitioned datasets, lineage-based recovery, and runtime query-plan adaptation. Its architecture supports low-latency execution and fault recovery across SQL and machine-learning operations.
- System architecture: Shark compiles Hive queries into RDD-based operator trees that Spark translates into tasks for execution on slave nodes.
- Spark: Spark provides general computation DAGs, in-memory RDD storage, automatic reconstruction after failures, and low-latency task management.
- RDDs: RDDs are immutable, partitioned datasets created by deterministic data-parallel operators, enabling lineage-based recomputation of lost partitions.
- Fault tolerance: Lineage recovery can rebuild lost partitions in parallel, tolerate worker loss, mitigate stragglers with speculative copies, and preserve recovery across combined SQL and machine-learning operations.
- SQL execution: Shark extends standard SQL planning with additional rule-based optimization and partial DAG execution that enables statistics-driven runtime re-optimization.
3 Engine Extensions
Shark extends Spark with runtime statistics and partial DAG execution to optimize distributed SQL, while adding storage and execution techniques that improve joins, reducers, and analytics workloads.
- Dynamic Query Optimization: Partial DAG execution lets Shark alter query plans at runtime using statistics collected during execution.It gathers global and per-partition statistics and can change operators or parameters such as parallelism.
- Join Optimization: Runtime join statistics let Shark select broadcast or shuffle strategies using exact input sizes, including for intermediate results without prior statistics.Broadcast joins avoid repartitioning and shuffling when one input is small.
- Runtime Parallelism: Shark uses partition sizes and greedy bin packing to choose reducer counts and mitigate skew at runtime.Fine-grained partitions are coalesced into fewer reducer partitions, with sizes approximately equalized when good packings exist.
- Runtime Parallelism: Shark can achieve similar performance improvement with more reduce tasks because Spark has low scheduling overhead.This contrasts with Hive, whose performance is especially sensitive to reducer counts because Hadoop has high scheduling overhead.
- Storage and Loading: Columnar in-memory storage reduces data size and processing time by as much as 5× over naïve Spark storage.Columnar representation also improves cache behavior for aggregation-heavy analytical queries, while JVM-object storage causes space overhead and costly garbage collection.
- Storage and Loading: 5× faster loading into memory than Hadoop was achieved while matching Hadoop’s HDFS loading throughput.The comparison used the same dataset as prior Hadoop-versus-MPP throughput experiments.
- Map Pruning: At least 3277 of 3833 sampled warehouse queries contained predicates usable for map pruning.Map pruning skips small memory-store blocks whose clustering-column values fall outside a query’s filter range.
4 Machine Learning Support
Shark unifies SQL and sophisticated machine learning by representing query results and operators as RDDs, allowing distributed analytics to share execution resources and cached data.
- Unified SQL and Machine Learning: Shark treats machine learning as a first-class workload alongside SQL by using Spark as its execution engine and RDDs as the main operator data structure.This design supports computation over query results without moving them out of the shared execution framework.
- Language Integration: Queries can return the RDD representing their plan, after which callers invoke distributed computation over the query result.This enables pipelines that combine SQL retrieval, feature extraction, and machine learning.
- Language Integration: Shark’s distributed logistic regression resembles single-node Scala code while combining SQL and MapReduce-style programming.Each iteration computes gradients across data points, reduces them to a net gradient, and updates the model vector.
- Execution Engine Integration: SQL can query machine-learning results within a single execution plan because both computations use RDDs.The same abstraction connects query operators and analytics computations.
- Execution Engine Integration: RDD-based operators let machine learning and SQL share workers and cached data without data-movement overhead.Because lineage spans the pipeline, failed machine-learning partitions can be recomputed automatically.
5 Implementation
Shark improves query-processing efficiency through engineering optimizations that reduce task, object, and data-structure overheads. It also identifies remaining opportunities for faster expression evaluation and specialized structures.
- Minimizing task tail latency and per-row CPU cost improves query-processing speed.
- Memory-based shuffle avoids filesystem and journaling overheads that increase shuffle variability.
- Minimizing temporary object creation reduces garbage-collection delays that can slow parallel jobs.
- Interpreted Hive expression evaluators waste most CPU cycles when data comes from memory.Shark’s planned JVM-bytecode compiler targets higher execution-engine throughput.
- Specialized data structures remain an unimplemented optimization that could compact primitive-key representations and improve cache behavior.
6 Experiments
Shark’s experiments evaluate SQL and machine-learning performance across benchmark, warehouse, and synthetic datasets. The results show large gains over Hive while preserving a MapReduce-like engine and fine-grained recovery features.
- Shark was evaluated on four datasets spanning Pavlo benchmarks, TPC-H, a real Hive warehouse, and machine learning.The datasets ranged from 100 GB synthetic data to 2.1 TB benchmark data and 1.7 TB warehouse data.
- 100× faster than Hive is Shark’s maximum reported SQL performance gain.
- Comparable performance to MPP databases was achieved while retaining a MapReduce-like engine and fine-grained fault recovery.The paper explicitly does not claim Shark is fundamentally faster than MPP databases.
- Six executions per query were run, with the first discarded and the remaining five averaged.The first run was discarded to allow JVM just-in-time compilation to optimize common code paths.
6.2 Pavlo et al. Benchmarks
On Pavlo et al.’s large-scale benchmarks, Shark substantially outperformed Hive across selection, aggregation, and join queries. Its advantages depended on workload characteristics, including memory placement, group cardinality, and co-partitioning.
- 6.2.1 Selection Query: 80× faster than Hive on in-memory data and 5× faster on HDFS data, Shark answered the selection query without a clustered index.
- 6.2.2 Aggregation Queries: Two million groups in one aggregation and approximately one thousand in the other exposed different scaling behavior across Shark, Hive, and MPP plans.Shark and Hive used task-local aggregation followed by a parallelized final merge.
- 6.2.2 Aggregation Queries: Shark outperformed Hive by a wide margin on both aggregation queries.
- 6.2.3 Join Query: Co-partitioning avoided shuffling 2.1 TB during the join and provided significant benefits.Serving join data from memory provided little benefit over disk because the join step dominated processing cost.
- 5× higher data-ingress rate was measured for Shark’s memory store than for HDFS when loading the 2 TB uservisits table.
6.3 Micro-Benchmarks
Micro-benchmarks evaluate Shark on TPC-H data, aggregation queries, runtime join selection, and failure recovery. Shark substantially outperforms Hive and improves execution through dynamic replanning and fine-grained recovery.
- Aggregation Performance: 80× faster than hand-tuned Hive for aggregation queries with small numbers of groups, and 20× faster for queries with large numbers of groups.For large-group queries, the shuffle phase dominated total execution cost.
- Aggregation Performance: Shark’s on-disk performance advantage was attributed to low task-launching overhead, an optimized shuffle operator, and other factors.Both Shark and Hive still read and deserialized data from HDFS in this comparison.
- Join Selection at Run-time: A dynamic optimizer selected a map-join after observing that filtering reduced 10 million suppliers to 1000, avoiding a costly shuffle join.Partial DAG execution exposed runtime statistics before the final join strategy was chosen.
- Join Selection at Run-time: 3× performance improvement over a naïve, statically chosen plan resulted from combining static query analysis with partial DAG execution.The optimizer pre-shuffled only the supplier table and avoided launching two waves of lineitem tasks.
- Fault Tolerance: Shark recovered from simulated node failures during group-by queries, and subsequent queries ran against the recovered dataset with fewer machines.Recovery was cheaper than reloading the entire dataset and re-executing the query; post-recovery performance was marginally better, likely due to JVM JIT effects.
6.4 Real Hive Warehouse Queries
Real Hive warehouse queries from a video analytics company test Shark on complex, high-dimensional data. Shark achieved sub-second latency in nearly all cases, while Hive took 50 to 100 times longer.
- Workload: The evaluated workload included summary statistics, filtered counts, distinct counts, grouping, sorting, and top-group selection.The queries operated across multiple dimensions and countries in the video-session warehouse.
- Real Workloads: Shark processed real warehouse queries in sub-second latency in all but one case, while Hive took 50 to 100 times longer.The workload contained 30 days of video session data occupying 1.7 TB decompressed.
- Real Workloads: Map pruning reduced the amount of data scanned by a factor of 30 on average.The queries’ data exhibited natural clustering properties.
6.5 Machine Learning
Shark integrates SQL-based data selection and feature extraction with iterative machine learning on large datasets. It outperformed Hive and Hadoop substantially, with lower gains for the more CPU-bound k-means workload.
- Workflow: Both machine learning algorithms ran for 10 iterations after SQL selection and feature extraction.The workflow consisted of warehouse filtering, feature extraction, and iterative algorithm execution.
- Performance: 100× faster than Hive and Hadoop for logistic regression, and 30× faster for k-means.The comparison measured the runtime of a single iteration using a 1-billion-row, 100 GB dataset on a 100-node cluster.
- Performance: K-means achieved less speedup because it was more computationally expensive than logistic regression and therefore more CPU-bound.Hadoop was evaluated with both text and serialized binary inputs; the binary format improved performance through compactness and lower deserialization cost.
- Workflow: When data was not already in Shark’s memory store, the first iteration took 40 seconds for both algorithms, while subsequent iterations matched the reported per-iteration results.Hive and Hadoop reloaded data from HDFS for every iteration.
7 Discussion
The discussion attributes Shark’s performance to in-memory and execution-engine improvements while preserving fine-grained task fault tolerance. It also identifies benefits of fine-grained tasks for skew, elasticity, and multitenancy, alongside an unexploited RDD capability.
- Fault Tolerance: Fine-grained task execution supports mid-query recovery by recomputing lost RDD data from lineage rather than replicating writes across the network.RDD operators are coarse-grained and deterministic, including map, group-by, and join.
- Sources of Performance: Shark combines in-memory storage, optimized shuffling, improved data layout, and partial DAG execution to alleviate MapReduce query bottlenecks.The discussion contrasts simple engineering changes with architectural changes such as partial DAG execution.
- Limitations: Shark does not yet exploit RDD random reads, which could allow RDDs to serve as indices and support remote-lookup joins.The limitation concerns read access; RDD writes remain coarse-grained.
- Task Scheduling Cost: Fine-grained tasks mitigate skew by distributing work across many short tasks, making imbalance less damaging to overall execution.The discussion contrasts this with Hadoop/Hive, where choosing the wrong number of tasks was sometimes 10× slower than optimal.
- Elasticity and Multitenancy: Fine-grained tasks allow nodes to join or leave during a query and support dynamic resource sharing between users.These properties provide elasticity and multitenancy benefits beyond query execution.
8 Related Work
Shark combines low-latency SQL, machine learning, and fine-grained fault recovery in one SQL engine, distinguishing it from prior system classes. Its design also avoids exporting SQL-selected data to another learning system.
- Shark is described as the only low-latency system combining SQL, machine learning workloads, and fine-grained fault recovery.
- Hive-like systems compile declarative queries into MapReduce-style jobs but struggle to achieve interactive response times.
- PowerDrill and Impala use shared-nothing parallel-database architectures but require whole-query re-execution after mid-query faults.
- HadoopDB and Osprey combine parallel databases with Hadoop or middleware, whereas Shark uses a simpler single-system architecture.
- Shark’s partial DAG execution complements single-node adaptive optimization by optimizing distributed shuffles while local tasks use traditional techniques.
- Unlike Spark, Shark integrates SQL-based data selection with immediate learning and provides more efficient relational in-memory representation plus mid-query optimization.
9 Conclusion
Shark combines fast relational queries and complex analytics in one fault-tolerant runtime. It uses database techniques and partial DAG execution to deliver large speedups while retaining machine learning and fine-grained mid-query recovery.
- Shark combines fast relational queries and complex analytics in a single fault-tolerant runtime.
- 100× faster than Hive for SQL and 100× faster than Hadoop for machine learning are the reported maximum speedups.
- 40–100× speedups on real queries were reported by two Internet companies using Shark as early users.