Chapter 6 ended with a promise: Spark’s third original goal — unifying real time and batch — is why the same engine later grows a streaming module and a SQL module. This chapter is that module, and it starts one step earlier, with the family of tools it belongs to.
SQL-on-Hadoop is a class of application tools that combine established SQL-style querying with newer Hadoop data framework elements. The two ingredients exist for two different reasons:
Every tool in this chapter is a wager that the common case matters more than the general case: most jobs are selections, projections, aggregations and joins over structured data, and a declarative statement of those jobs is enough for an optimiser to do the hard work. The RDD remains the escape hatch for everything else — the deck puts it as “whereas RDDs enable any computation through user defined functions”. SQL-on-Hadoop never claims to replace MapReduce; it claims that most computations do not need MapReduce’s full power.
SQL-on-Hadoop is not one thing. The deck splits the tools into three categories, and the split is worth memorising because it predicts everything else about a tool — latency, storage, engine, and how resources are allocated:
Examples: Apache Hive, Apache Spark.
SQL(-like) queries are translated into MapReduce/Spark jobs; they run on very large datasets and large ETL jobs, with latencies of minutes to hours. The Spark flavour adds in-memory SQL: load the datasets in memory, then query them.
Examples: Impala, Apache Drill.
Interactive queries to enable traditional BI and analytics, with latencies of milliseconds to minutes. They require always-on daemons — the price of interactivity is that the engines are running before any query arrives.
Examples: Apache HBase, NoSQL databases.
OLTP workloads and web applications that operate over smaller datasets, typically with insert, update, and deletes at millisecond latency. Note that HBase’s SQL support is limited compared to the other families — it earns its place in the table through the query workload, not through full SQL.
The deck condenses all of this into one comparison table. Read it as a table of responsibilities: who stores the data, who executes the queries, and who gets blamed for the resources:
| Category | Query tool | Data storage | Execution engine | Allocation of resources |
|---|---|---|---|---|
| Operational | NoSQL | Own | Own | Per service |
| Operational | HBase | DFS | Own | Per service |
| Interactive | Impala | DFS | Own (MPP-style) | Per service |
| Interactive | Drill | Any | Own (MPP-style) | Per service |
| Batch | Spark | Any | Own (YARN) | Per job |
| Batch | Hive | DFS | MapReduce, Tez, Spark | Per job |
The two cells that break the pattern are Hive (batch, storage on DFS, but no engine of its own) and Spark (batch, any storage, own engine on YARN). Both are allocated per job — that is the family trait of batch SQL, and it is why the interactive tools, with their always-on daemons, cost resources even when idle.
Apache Hive is the oldest and most influential member of the family: a data warehouse software that facilitates reading, writing, and managing large datasets residing in distributed storage using SQL.
Three properties define how it works:
Hive’s best use is batch jobs over large sets of append-only data. It is explicitly not for real-time queries (high latency) and not designed for OLTP. When the slides say “data warehouse” they mean the analytic kind: many reads, few writes, no interactive expectations. This is the same boundary the interactive family exists to serve.
And the ecosystem note worth remembering: AWS Athena enables SQL querying over S3 data, relies on Apache Hive for data modeling, and relies on Presto (an MPP-like tool) to run queries — a managed service standing on the shoulders of the very tools this chapter describes.
Hive organises data in four nested units, and each unit is practically a folder (or a set of files) in HDFS. The deck gives the exact paths, so the picture is worth drawing once:
| Unit | What it is | HDFS location |
|---|---|---|
| Database | Container of tables | /user/hive/warehouse/dbname.db |
| Table | Homogeneous set of records with the same schema | /user/hive/warehouse/dbname.db/tablename/* |
| Partition (optional) | Records split by the value of one or more columns — a further level of folders | .../tablename/column=value/* |
| Bucket (optional) | Records grouped into a fixed number of files, based on the hash value of some column — one file per bucket | .../tablename/bucket01 |
The cardinality rule is the examinable nugget. Partition on low-cardinality columns (a handful of distinct values, e.g., departments) because each value is worth a whole folder; partitioning on a high-cardinality column would create thousands of near-empty folders — the “explosion” the slides warn about. Bucket on high-cardinality columns (e.g., IDs) because hashing spreads records evenly across a fixed number of files, and bucketing columns give faster search and a skipped shuffle on joins — a first taste of Chapter 8’s partitioning logic, at the storage layer.
Hive showed that structure can be bolted onto Hadoop. Spark SQL does it inside the engine itself. The definition in the deck is one sentence long, and it is a direct continuation of Chapter 6:
RDDs are immutable distributed collection of objects. DataFrames and DataSets are immutable distributed collection of records organized into named columns (i.e., a table). Simply put: RDDs with a schema attached.
Everything else about them follows from that one sentence:
The one difference between the two APIs is when type conformity is checked:
Records organized into named columns, with the schema checked at runtime. This is the more permissive, dynamically-typed feel — and the slide explicitly connects it to Chapter 6’s warning: type inference is a Scala feature and is different from dynamic typing.
The same collection of records with named columns, but type conformity is checked at compile time (the DataSet is the typed API). Checking earlier avoids querying columns that do not really exist — the error is caught before the job is even submitted, which is exactly the “anticipate programming errors” benefit of Chapter 6’s type inference.
And then the sentence that carries the chapter: they are still lazily evaluated — nothing from Chapter 6 is thrown away — but they support under-the-hood optimizations and code generation:
Three one-line distinctions to keep ready: RDD = collection of objects, DataFrame/DataSet = collection of records with named columns (an RDD with a schema); DataFrame checks types at runtime, DataSet at compile time; RDDs are lazily evaluated and so are DataFrames/DataSets — but the latter get Catalyst optimizations and JVM code generation on top. The laziness is the same laziness; the difference is that the schema gives the optimiser something to work with.
The deck stops to justify the whole design, and the honesty is refreshing: structure is a trade, not a gift. The slides put it as a balanced list:
| Argument | |
|---|---|
| Cons | Structure imposes some limits — whereas RDDs enable any computation through user defined functions, a schema only fits computations that can be expressed on top of it. |
| Pros | The most common computations are supported — selections, aggregations, joins cover the majority of real workloads. |
| Language simplicity — SQL is widely understood, so analysts and developers share one language. | |
| It opens the room to optimizations — it is hard to optimize a user defined function, because the optimiser cannot see inside it; a declarative plan, it can. |
The cons column is exactly why RDDs still exist, and the pros column is exactly why Catalyst can exist. A user defined function is a black box: Spark cannot reorder it, prune columns through it, or push predicates into it. A relational plan is a tree of operators with known semantics, and a tree can be rewritten. Structure is not a restriction the user suffers; it is the information the optimiser needs. Chapter 6’s laziness is what makes the rewriting possible — the plan exists before any data moves.
The Catalyst optimizer creates optimized execution plans by working on two distinct representations of the query, and the deck draws the boundary sharply:
| Plan | It describes |
|---|---|
| Logical Plan | Which computations must be done — the relational operations, independent of any algorithm. |
| Physical Plan | Which computations must be done and how to conduct them — i.e., which algorithms are used. |
This is the same split as Chapter 6 at one level up: the logical DAG of RDDs became a physical plan of stages and tasks; here, the logical query plan becomes a physical plan of algorithms. The whole journey of a query, from text to execution:
Logical optimization is based on rules: a rule is a function that can be applied on a portion of the logical plan, implemented as a Scala function. The deck lists several types of rules:
| Rule | What it does |
|---|---|
| Constant folding | Resolve constant expressions at compile time — e.g., WHERE price > 10 AND 1 = 1 is simplified before any data is read. |
| Predicate pushdown | Push selection predicates close to the sources — the filter happens as early (and as cheaply) as possible, ideally during the physical scan. |
| Column pruning | Project only the required column — drop every column the query never uses. |
| Join reordering | Change the order of join operations — joining small inputs first beats joining large ones first. |
The rules are applied recursively and iteratively until the plan reaches a fixed point — run them until none of them changes anything anymore. Each application may enable another: pruning columns makes a join cheaper, which may make reordering it worthwhile, and so on.
Select each line to see what Catalyst does at that step. The SQL is the same query from the exam material — only the plan underneath it changes.
None of these rules could exist on plain RDDs, because an RDD carries no information about what its elements mean. The moment records have named columns, a rule can ask “is this predicate applicable here?” and answer with the schema. Constant folding, predicate pushdown, column pruning and join reordering are four small rewrites, but they are applied until fixed point — and the combination is where the real gains come from, exactly as with Chapter 6’s pipelining of narrow dependencies.
Once the logical plan is optimized, one or more physical execution plans are defined — each a concrete choice of algorithms — and a cost model is used to select the best one. The model in the slides is deliberately simple:
Cost(table) = α · costCPU + (1−α) · costIO
= α · cardinality + (1−α) · sizeTwo components, one knob: costCPU is the cardinality (how many records), costIO is the size (how many bytes), and α decides how much each matters. On top of the cost model there are also rule-based physical optimizations:
Slide α and the two components, and watch the cost table move. The model only ever sees these two numbers — that is its strength and, as section 11 shows, its blind spot.
The most visible effect of physical optimization is join method selection: the same join can be executed by three different algorithms, and the cost model picks one. All three are worth drawing because the exam routinely asks for their conditions:
Formally, the deck defines them as:
The selection is a small decision tree, and the conditions are exact:
spark.sql.autoBroadcastJoinThreshold, default 10MB (some people increase it to GB levels).The deck closes with a caveat that matters for the exam’s tone: join methods are under continuous refinement, and a few more exist for certain situations (e.g., cartesian product).
Toggle the three conditions and apply the decision tree. The verdict is what Catalyst’s cost model would conclude.
The three conditions, verbatim: Broadcast if one RDD is smaller than the threshold (autoBroadcastJoinThreshold, default 10MB); Shuffle Hash if Broadcast does not apply and one RDD is 3× smaller and the smaller partitions fit in memory as hash tables; Shuffle Sort Merge otherwise. Notice the logic is exclusion-based: Broadcast first, then Shuffle Hash, then Sort Merge as the residual case — the same “conditions that do not apply” phrasing appears in the slides, and the exam mirrors it.
The deck is candid about the model’s limits: Catalyst only considers the size and the cardinality of tables. Everything else that determines a query’s real cost is missing from the formula:
A cost model is only useful if it can be evaluated before the query runs, with information the optimiser already has. Cardinality and size are known at planning time; network and disk throughput are properties of a cluster that can change between jobs, and data locality is a scheduling decision made after the plan is fixed. The simplification is not ignorance — it is a choice of which costs are plannable. What the deck calls research work is precisely the attempt to make the model probabilistic and accurate anyway.
And the research is from the course’s own group — worth citing, because it is the visible frontier of this chapter’s topic:
Both attack the same gap: defining a more accurate, probabilistic cost model — one that turns Chapter 8’s tuning knobs into something the optimiser itself can reason about.
The chapter ends where the current version of Spark ends: Adaptive Query Execution (AQE), introduced with version 3.0. Its main idea in one sentence:
The execution plan is not final. Reviews are made at each stage boundary, and additional optimizations are possibly applied, given the information available on the intermediate data. AQE can be defined as a layer on top of Spark Catalyst which will modify the Spark plan on the fly.
Why does this make sense? The cost model of section 9 works on estimates. Once a stage has actually run, Spark knows the real cardinalities and sizes — so the plan can be revised with facts instead of guesses. The deck lists the two drawbacks with equal honesty:
And four concrete optimizations, each fixing one specific planning blind spot:
| AQE optimization | What it does |
|---|---|
| Adaptive number of shuffle partitions | Spark SQL used to set a default number of 200 partitions at each stage; AQE automatically adjusts it at runtime based on the actual data volume. |
| Dynamically converting Sort Merge joins to Broadcast joins | A dynamic switch of join strategies based on actual table sizes — a table estimated as large but actually small becomes a Broadcast join mid-flight. |
| Dynamically coalesce shuffle partitions | Merge the partitions that ended up nearly empty, so the next stage does not schedule tiny tasks. |
| Dynamically optimize skewed joins | Split the partitions dominated by a single popular key, so one straggler no longer dictates the join’s runtime. |
Step through a two-stage query and watch the plan change at the stage boundary. Each review uses data the planner could only guess before.
AQE is not a fourth join method and not a new optimizer — it is a layer on top of Catalyst that re-runs parts of the optimization with real statistics. And its drawback list is examinable too: the stop-at-each-boundary pause, and the plan that is harder to read because the original multi-stage job becomes a set of single-stage jobs.
Chapter 6 said stages’ boundaries are defined by shuffle operations; AQE is exactly what makes those boundaries useful: they are the points where the engine may stop, look at real intermediate data, and re-plan. If you are asked why AQE exists, answer in two clauses: the cost model plans from estimates, and the stage boundary is where the estimates become facts.
SQL-on-Hadoop is a class of application tools that combine established SQL-style querying with newer Hadoop data framework elements. It exists because MapReduce-style programming is the native and most comprehensive way to design jobs, but SQL is a simple and widely understood language — so most computations can be expressed declaratively and let an optimiser do the hard work, while RDDs remain the escape hatch for everything a schema cannot express.
Batch SQL (Apache Hive, Apache Spark): SQL(-like) queries translated into MapReduce/Spark jobs, on very large datasets and large ETL jobs, minutes to hours; Spark adds in-memory SQL (load datasets in memory, then query them). Interactive SQL (Impala, Apache Drill): interactive queries for traditional BI and analytics, milliseconds to minutes, requiring always-on daemons. Operational SQL (Apache HBase, NoSQL databases): OLTP workloads and web applications over smaller datasets, typically with insert, update and deletes at millisecond latency.
Hive breaks the execution-engine column: it stores on DFS but pushes execution to other tools (MapReduce, Tez, Spark), being the only engine-less tool. Spark breaks the storage column: it can read/write from/to any source and runs its own engine on YARN. Both are batch tools with per-job allocation of resources, unlike the interactive/operational families which are allocated per service.
Hive is a data warehouse software that facilitates reading, writing, and managing large datasets residing in distributed storage using SQL. Structure is projected onto data already in storage (external tables) through metadata, and that metadata is stored in a Metastore using an RDBMS on a single node — a single point of failure, although High-Availability is supported. Data is queried via HiveQL (a subset of SQL), queries are translated to MapReduce/Spark jobs, and a procedural language (HPL-SQL) is supported.
Best use: batch jobs over large sets of append-only data. It is not for real-time queries (high latency) and not designed for OLTP.
Database: container of tables, practically a folder — /user/hive/warehouse/dbname.db. Table: homogeneous set of records with the same schema, stored as a set of files — .../tablename/*. Partition (optional): records split by the value of one or more columns, implemented as a further level of folders — .../tablename/column=value/*. Bucket (optional): records grouped into a fixed number of files based on the hash value of some column — .../tablename/bucket01.
Partition on low-cardinality columns (e.g., departments): each value deserves a folder, and search on partitioning columns is faster — but there is a potential “explosion” of folders if cardinality is high. Bucket on high-cardinality columns (e.g., IDs): hashing spreads records evenly across a fixed number of files, giving faster search and a skipped shuffle on bucketing columns.
RDDs are immutable distributed collections of objects; DataFrames and DataSets are immutable distributed collections of records organized into named columns (i.e., a table) — simply put, RDDs with a schema attached. They support both relational and procedural processing (e.g., SQL and Scala), complex data types (struct, array) and user defined types, are cached using columnar storage, and can be built from DBMSs, files, other tools (e.g., Hive) and RDDs.
At compile time for DataSets; at runtime for DataFrames. Checking earlier — as the DataSet API does — avoids querying columns that do not really exist, catching the error before the job is submitted.
Catalyst creates optimized execution plans for the still-lazily-evaluated DataFrames/DataSets: IO optimizations such as skipping blocks in Parquet files and logical push-down of selection predicates. JVM code generation produces code for all supported languages, even non-native JVM ones such as Python, and consists in clubbing multiple physical operations together to form a single Java function — so records flow through fused operations without interpreter round-trips.
Con: structure imposes some limits — whereas RDDs enable any computation through user defined functions. Pros: the most common computations are supported; language simplicity; and structure opens the room to optimizations — it is hard to optimize a user defined function, because the optimiser cannot see inside it.
The logical plan describes which computations must be done; the physical plan describes which computations must be done and how to conduct them (i.e., which algorithms are used). The four logical rules: constant folding (resolve constant expressions at compile time), predicate pushdown (push selection predicates close to the sources), column pruning (project only the required column), join reordering (change the order of join operations). Rules are implemented as Scala functions, applied recursively and iteratively until the plan reaches a fixed point.
Cost(table) = α·costCPU + (1−α)·costIO = α·cardinality + (1−α)·size. The model only considers the size and the cardinality of tables; it ignores network throughput, disk throughput, the allocation of resources (number of executors, cores per executor) and the allocation of tasks (data locality probability). Rule-based physical optimizations (operation pipelining, predicate push-down during the physical scan) run alongside it.
Broadcast Hash join: the smaller table is fully loaded into each executor’s memory, similar to broadcast variables — selected when one RDD is smaller than the threshold spark.sql.autoBroadcastJoinThreshold (default 10MB). Shuffle Hash join: both tables repartitioned by join key; the hash table of the smaller partition is loaded into memory and the bigger partition probes it — selected when Broadcast does not apply, one RDD is 3 times smaller than the other, and the smaller partitions fit in memory as hash tables. Shuffle Sort Merge join: both tables repartitioned by join key, keys sorted within each partition, join by iterating on the sorted lists — selected when neither of the previous conditions applies.
Adaptive Query Execution, introduced with Spark 3.0: the execution plan is not final; reviews are made at each stage boundary and additional optimizations are applied given the information on the intermediate data. It is a layer on top of Catalyst that modifies the Spark plan on the fly. Optimizations: adaptive number of shuffle partitions (the default 200 per stage is adjusted at runtime), dynamically converting Sort Merge joins to Broadcast joins (a dynamic switch of join strategies based on actual table sizes), dynamically coalescing shuffle partitions, and dynamically optimizing skewed joins. Drawbacks: execution stops at each stage boundary for the review (usually worth it), and the execution plan is harder to read — the original multi-stage job becomes a set of single-stage jobs.