Joins, Deduplication, Aggregation and Spark Tuning

What it is

Silver tables are combined and summarized to answer business questions. Joins match rows across DataFrames by key, unions stack DataFrames with the same columns, deduplication removes repeated records and aggregations compute counts and averages.

Joins and aggregations shuffle data across the cluster, so a few Spark settings control how fast they run. The exam expects you to know what each setting changes and to re-measure after changing it.

How it works

Join types

from pyspark.sql import functions as F
result = (orders
  .join(customers, ["customer_id", "region"], "left")
  .join(F.broadcast(countries), "country_code"))
pairs = sizes.crossJoin(colors)

Passing a list of column names joins on multiple keys and keeps one copy of each key column. A condition such as orders.id == customers.id keeps both columns, which then need dropping or renaming. In SQL, /*+ BROADCAST(c) */ is the broadcast hint.

A broadcast join avoids shuffling the large table. It fits when one side is small enough to fit in executor memory, such as a lookup table.

Union and union all

PySpark union behaves like SQL UNION ALL. Add .distinct() to remove duplicates, and use unionByName when the column order differs between the DataFrames.

Deduplication

distinct() removes rows that are identical in every column. dropDuplicates(["order_id"]) keeps one row per key and ignores the other columns; which row survives is not guaranteed.

To keep a specific row, such as the latest version, rank rows in a window and keep the first:

from pyspark.sql import Window, functions as F
w = Window.partitionBy("order_id").orderBy(F.col("updated_at").desc())
latest = (orders.withColumn("rn", F.row_number().over(w))
  .filter("rn = 1").drop("rn"))

Aggregations

from pyspark.sql import functions as F
stats = orders.groupBy("region").agg(
  F.count("*").alias("orders"),
  F.approx_count_distinct("customer_id").alias("customers"),
  F.mean("amount").alias("avg_amount"))

describe() gives count, mean, stddev, min and max; summary() adds the 25%, 50% and 75% percentiles. In SQL the function is also approx_count_distinct.

Basic tuning parameters

Session settings such as shuffle partitions change at runtime with spark.conf.set. Memory settings are fixed when the cluster starts, so they go in the cluster's Spark config. Serverless compute allows only a few properties, including spark.sql.shuffle.partitions, and manages memory itself.

spark.conf.set("spark.sql.shuffle.partitions", "64")
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 50 * 1024 * 1024)

Adaptive query execution, on by default, can coalesce shuffle partitions and switch to a broadcast join at runtime. Change one setting at a time and re-measure with the same data, comparing duration and stage metrics in the Spark UI.

How the exam asks about this

• "Join a 2 TB fact table to a 5 MB country lookup without shuffling the fact table." Use a broadcast join.
• "Stack two monthly DataFrames and keep every row, columns in a different order." Use unionByName.
• "Count unique visitors on billions of rows, where an estimate is acceptable and speed matters." Use approx_count_distinct.