Joins and Aggregations in Spark SQL and PySpark: Join Types, Grouping Sets and Scale

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.

-- 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      6

USING 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.

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.

query = (streamingDF
  .join(staticDF, "user_id", " ? ")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(availableNow=True)
  .table(f"{catalog}.{schema}.bookings_with_user_info")
)

Sources23

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.

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|

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.

Sources5

Exam traps

Each one states something that sounds right. Open it to see what is actually true.

1. 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. 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. 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