What you will be able to do
- Choose the join type (inner, outer, semi, anti, cross) that returns the rows a task needs, in SQL and with DataFrame.join
- Explain how batch, stream-static and stream-stream joins differ in state and results
- Write multi-level aggregations with GROUPING SETS, ROLLUP and CUBE, and count the grouping sets each one produces
- Choose between exact batch aggregation, materialized views and approximate aggregate functions for large data
1.Join types and what each returns
Databricks supports ANSI join syntax: [ join_type ] JOIN right_table_reference [ join_criteria ]. The criteria can be ON boolean_expression, or USING (col, ...) when the column names match on both sides. If you omit the join type, you get an INNER join. A NATURAL join matches automatically on all columns with the same name. It can't be combined with CROSS or LATERAL, and trying raises INCOMPATIBLE_JOIN_TYPES.
The join type decides which rows survive. Semi and anti joins are the ones people often forget. They return only left-side columns, and they act as "exists" and "not exists" filters, which makes them the natural way to find orphan or unmatched records.
| Join type | Returns |
|---|---|
| [ INNER ] | Rows with matching values in both sides (the default) |
| LEFT [ OUTER ] | All left rows, plus matched right values or NULL |
| RIGHT [ OUTER ] | All right rows, plus matched left values or NULL |
| FULL [ OUTER ] | All rows from both sides, NULL where there is no match |
| [ LEFT ] SEMI | Left rows that have a match on the right |
| [ LEFT ] ANTI | Left rows that have no match on the right |
| CROSS JOIN | Cartesian product of the two relations |
-- Use employee and department tables to demonstrate anti join.
> SELECT *
FROM employee
ANTI JOIN department ON employee.deptno = department.deptno;
105 Chloe 5
104 Evan 4
106 Amy 6USING also changes the shape of the output. With USING or NATURAL, SELECT * shows each join column once, first in the list, followed by the remaining left columns and then the remaining right columns. An ON join keeps both copies of the key, which is where AMBIGUOUS_REFERENCE errors come from.
Checkpoint 1 of 6· Match them up
Match each requirement to the join type that meets it
Tap a term, then the definition that fits it.
Semi and anti joins filter the left side by whether a match exists. Outer joins keep unmatched rows and fill the gaps with NULL.
“Returns the values from the left table reference that have no match with the right table reference.”Source: docs.databricks.com
Sources1
2.DataFrame joins, and joins over streams
In PySpark, the same choices go through DataFrame.join(other, on, how). on can be a column name, a list of names, or a join expression (Column). If you pass names, the columns must exist on both sides, and you get an equi-join with a single copy of each key, like SQL USING. how defaults to inner and accepts these values: inner, cross, outer/full/full_outer, left/left_outer, right/right_outer, semi/left_semi and anti/left_anti.
df.join(df2, "name").show()
# +----+---+------+
# |name|age|height|
# +----+---+------+
# | Bob| 5| 85|
# +----+---+------+On large pipelines, where the data comes from matters as much as the syntax. Batch joins are stateless. Each run recomputes results from the source data as it is at that moment. Stream-static joins are also stateless. Each micro-batch is joined to the latest valid version of a Delta table, so you don't need a watermark. This is the standard way to enrich a fact stream with a slowly changing dimension. There's a catch, though. If the static table changes between runs, reprocessing the same stream can give different results. Stream-stream joins are stateful. Databricks recommends watermarks on both sides, and the supported types are inner, left/right/full outer and left semi. Watermark mechanics belong to the streaming lesson.
On performance, Databricks says skew-join hints aren't needed because it optimizes skewed joins automatically. Range-join hints can help with inequality joins, for example on timestamp ranges.
Checkpoint 2 of 6· Fill the gap
This stream-static join enriches each booking with the user's name and country, and drops bookings that have no matching user. Which join type completes it?
query = (streamingDF
.join(staticDF, "user_id", " ? ")
.writeStream
.option("checkpointLocation", checkpoint_path)
.trigger(availableNow=True)
.table(f"{catalog}.{schema}.bookings_with_user_info")
)An inner join keeps only bookings whose user_id matches the static users table. left_anti would keep only the unmatched bookings, and cross would pair every booking with every user.
Source: docs.databricks.comCheckpoint 3 of 6· Exam question
A batch job joins a 40 GB `orders` Delta table with a 6 MB `region_lookup` Delta table on `region_id`, using a plain PySpark `.join()` call with no explicit hints. Adaptive query execution (AQE) is enabled with default settings. Which statement correctly describes what happens at execution time?
Correct answer: A — AQE can convert the join to a broadcast hash join after the shuffle stage measures `region_lookup`'s actual runtime size and finds it under the broadcast threshold, even without a `broadcast()` hint in the code.
- A. Correct. One of AQE's core runtime reoptimizations is converting sort-merge join to broadcast join once actual post-shuffle statistics show one side is small enough, so a 6 MB lookup table can be broadcast dynamically even without a code hint.
- B. Incorrect. AQE's documented capabilities go beyond skew handling and partition coalescing; dynamically switching a sort-merge join to a broadcast join based on runtime statistics is one of its three main features.
- C. Incorrect. Broadcast joins do not require an explicit hint; Spark's static optimizer and AQE can both select a broadcast strategy automatically based on estimated or measured table size relative to the broadcast threshold.
- D. Incorrect. Broadcast eligibility in Spark is governed by the estimated or measured byte size of the smaller side relative to `spark.sql.autoBroadcastJoinThreshold`, not by a distinct row-count threshold that triggers a nested loop strategy.
3.Multi-level aggregation: GROUP BY, GROUPING SETS, ROLLUP, CUBE
GROUP BY collapses rows into groups and applies aggregate functions such as MIN, MAX, COUNT, SUM and AVG. Three features make it shorter to write:
- GROUP BY ALL (DBR 12.2 LTS and above) groups by every SELECT-list expression that doesn't contain an aggregate.
- DISTINCT inside an aggregate removes duplicate inputs before aggregating.
- FILTER (WHERE ...) on an aggregate passes only the matching rows to that one function. You can compute several conditional metrics in one pass.
Reports often need subtotals and a grand total along with the detail rows. GROUPING SETS returns several groupings in one statement and works like a UNION ALL of separate GROUP BY queries. The empty set () is the global aggregate.
| Clause | Equivalent GROUPING SETS | Sets for N columns |
|---|---|---|
| ROLLUP(warehouse, product) | ((warehouse, product), (warehouse), ()) | N+1 |
| CUBE(warehouse, product) | ((warehouse, product), (warehouse), (product), ()) | 2^N |
| GROUPING SETS((warehouse), (product)) | Union of GROUP BY warehouse and GROUP BY product | As listed |
ROLLUP is hierarchical. It drops columns from the right, which suits city → model style drill-downs. CUBE covers every combination. PySpark has the same methods, rollup, cube and groupingSets, plus groupBy(...).agg(...) and pivot. In the result, a NULL in a grouping column marks a subtotal row for that level:
df.groupingSets(
[("city", "car_model"), ("city",), ()],
"city", "car_model"
).agg(sf.sum(sf.col("quantity")).alias("sum")).sort("city", "car_model").show()
# +--------+------------+---+
# | city| car_model|sum|
# +--------+------------+---+
# | NULL| NULL| 78|
# | Dublin| NULL| 33|
# | Dublin|Honda Accord| 10|Checkpoint 4 of 6· Check yourself
Which clause is equivalent to GROUP BY GROUPING SETS((warehouse, product), (warehouse), ())?
ROLLUP over two columns gives N+1 = 3 sets, dropping columns from the right. CUBE would add a fourth set, (product).
“The N elements of a ROLLUP specification result in N+1 GROUPING SETS.”Source: docs.databricks.com
Sources4
4.Aggregating large datasets: batch, incremental, streaming, approximate
A batch aggregate, whether an ad hoc SQL query or a DataFrame groupBy, scans every record in the source each time it runs. Databricks optimizes these queries with metadata, but latency and cost still grow with data size. Databricks gives three answers to that:
1. Materialized views for aggregates you query often. They track changes in the source and update the aggregates on refresh, and the results match a full recomputation.
2. Streaming aggregates are stateful, and they must use a watermark. Without one, state grows without limit, which can lead to out-of-memory errors. Databricks also says not to use a stateful aggregate for statistics over an entire dataset, and to use a materialized view instead.
3. Approximation, when an exact answer isn't needed. Spark SQL provides approx_count_distinct, approx_percentile and approx_top_k, and you can run aggregates over a random TABLESAMPLE. LIMIT isn't a valid substitute for sampling, because it doesn't introduce randomness.
Checkpoint 5 of 6· Check yourself
An analyst needs a fast, cheap estimate of distinct users across a multi-billion-row table and doesn't need an exact figure. Which approach fits Databricks guidance?
Approximate aggregate functions trade a little precision for speed and cost. LIMIT isn't random and doesn't spread its sample across the dataset.
“using approximation for aggregates can accelerate query processing and reduce costs when you don't require precise results.”Source: docs.databricks.com
Checkpoint 6 of 6· Exam question
A Spark SQL job joins a heavily skewed `events` table (one `user_id` accounts for 30% of all rows) to a `users` table on `user_id`. The Spark UI shows one task in the join stage running far longer than the rest, stalling the whole stage. AQE is already enabled. Which two actions would most directly address this specific skew, without changing the business logic of the join? (Select 2)(Select 2)
Correct answers: A, B — Let AQE's skew join optimization split the oversized `user_id` partition into several smaller sub-partitions and replicate the matching `users` rows so each sub-partition can be processed by a separate task.; Manually salt the join key on both sides by appending a random integer suffix to `user_id`, exploding the matching side across the salt range, and joining on the composite salted key.
- A. Correct. AQE skew join optimization detects an oversized post-shuffle partition, splits it into multiple sub-partitions, and replicates the corresponding rows from the other side so the work for that key is spread across parallel tasks instead of one long-running task.
- B. Correct. Salting distributes a single hot key across many synthetic keys on both sides of the join, breaking up the oversized partition into several smaller ones that can be processed in parallel, which is the classic manual fix when automatic skew handling is insufficient.
- C. Incorrect for this scenario. Broadcasting the small `users` table avoids a shuffle of `users` but `events` can still be scanned and processed with data-local partitioning that leaves the hot `user_id` rows concentrated; broadcast joins are also explicitly not skew-optimized by AQE.
- D. Incorrect. Increasing the number of default shuffle partitions changes how many buckets hash partitioning uses, but every row for the same `user_id` still hashes to the same single partition, so the oversized partition for that key remains oversized.
- E. Incorrect. Caching changes where bytes are read from but does not change how rows are partitioned across tasks for the join, so the one partition holding 30% of the key's rows still overwhelms a single task.
Sources5
Exam traps
Each one states something that sounds right. Open it to see what is actually true.
1.Leaving out the ON clause makes a LEFT JOIN behave like a lookup that returns NULLs.Why is that wrong?
Without join criteria, every join type becomes a CROSS JOIN and returns the Cartesian product.
Covered in Join types and what each returns
2.A stream-static join always gives the same output when the same stream data is reprocessed.Why is that wrong?
Each micro-batch joins against the latest version of the static table, so updating that table between runs can change the results.
Covered in DataFrame joins, and joins over streams
3.Running a streaming (stateful) aggregate is a good way to keep a whole-table statistic up to date.Why is that wrong?
Databricks advises against stateful aggregates over an entire dataset and recommends materialized views for incremental whole-dataset aggregation.
Covered in Aggregating large datasets: batch, incremental, streaming, approximate
Sources
Every claim above is drawn from one of these pages, quoted as it was written on the date shown.
- 1.
“Returns values from the left side of the table reference that has a match with the right.”
↩︎ Join types and what each returns“SELECT * will only show one occurrence for each of the columns used to match first”
↩︎ Join types and what each returns“If you omit the join_criteria the semantic of any join_type becomes that of a CROSS JOIN.”
↩︎ Exam trap 1“If you omit the join_criteria the semantic of any join_type becomes that of a CROSS JOIN.”
↩︎ Prediction“Returns the values from the left table reference that have no match with the right table reference.”
↩︎ Checkpoint - 2.
“the column(s) must exist on both sides, and this performs an equi-join.”
↩︎ DataFrame joins, and joins over streams - 3.https://docs.databricks.com/aws/en/transform/joinOfficial docs
“All batch joins are stateless joins.”
↩︎ DataFrame joins, and joins over streams“Databricks recommends specifying watermarks for both sides of all stream-stream joins.”
↩︎ DataFrame joins, and joins over streams“Hints for skew joins aren't necessary because Databricks automatically optimizes these joins.”
↩︎ DataFrame joins, and joins over streams“If you update the static table between runs, re-processing the same streaming data can produce different results.”
↩︎ Exam trap 2 - 4.https://docs.databricks.com/aws/en/sql/language-manual/sql-ref-syntax-qry-select-groupbyOfficial docs
“When a FILTER clause is attached to an aggregate function, only the matching rows are passed to that function.”
↩︎ Multi-level aggregation: GROUP BY, GROUPING SETS, ROLLUP, CUBE“A shorthand notation to add all SELECT-list expressions not containing aggregate functions as group_expressions.”
↩︎ Multi-level aggregation: GROUP BY, GROUPING SETS, ROLLUP, CUBE“The N elements of a CUBE specification results in 2^N GROUPING SETS.”
↩︎ Prediction“The N elements of a ROLLUP specification result in N+1 GROUPING SETS.”
↩︎ Checkpoint - 5.
“Databricks recommends using materialized views to incrementally update aggregate values.”
↩︎ Aggregating large datasets: batch, incremental, streaming, approximate“You must use watermarks when computing stateful aggregates.”
↩︎ Aggregating large datasets: batch, incremental, streaming, approximate“does not introduce randomness, or guarantee that sampling is distributed across the dataset.”
↩︎ Aggregating large datasets: batch, incremental, streaming, approximate“You should not use a stateful aggregate to calculate statistics over an entire dataset.”
↩︎ Exam trap 3“using approximation for aggregates can accelerate query processing and reduce costs when you don't require precise results.”
↩︎ Checkpoint