Skip to content
FR

Deduplication, Aggregations, and Basic Spark Tuning

Step 13 of 30

Deduplication, Aggregations, and Basic Spark Tuning

Introduction

This lesson continues the Data Transformation and Modeling section of the official exam outline (about 22% of the scored questions). In lesson 6 you cleaned and combined data. Now you remove duplicates, compute summary statistics, and learn the handful of Spark settings that the exam expects you to recognize: shuffle partitions, parallelism, memory and the broadcast threshold. The exam does not ask you to be a performance engineer. It asks whether you know which setting matches which symptom, and whether you measure before and after a change.

After this lesson you will be able to choose between dropDuplicates and a window function, compute aggregates and profiles with count, approx_count_distinct, mean, summary and describe, and name the effect of the main Spark configuration properties.

Learning objectives

  • Remove exact and key-based duplicates with distinct, dropDuplicates and ROW_NUMBER() windows, and keep the latest record per key.
  • Compute aggregates with groupBy/GROUP BY, and explain how null values affect count.
  • Use approx_count_distinct, describe and summary for fast data profiling.
  • Explain the roles of spark.sql.shuffle.partitions, spark.default.parallelism, spark.executor.memory, spark.driver.memory and spark.sql.autoBroadcastJoinThreshold.
  • Describe how adaptive query execution (AQE) and auto-optimized shuffle reduce manual tuning.
  • Apply a disciplined measure, change one setting, re-measure workflow.

1. Deduplication

Duplicates appear whenever sources retry, files are re-delivered, or change feeds contain several versions of a record. There are two different problems, and the exam checks that you can tell them apart.

Problem Meaning Tool
Exact duplicates Entire rows are identical distinct(), dropDuplicates() with no argument, SELECT DISTINCT *
Key duplicates Several rows share a business key but differ in other columns (for example an older and a newer version) dropDuplicates(["key"]) (arbitrary row kept) or a window function (you choose which row is kept)

distinct and dropDuplicates

exact = orders.dropDuplicates()                    # same as distinct()
by_key = orders.dropDuplicates(["order_id"])       # one row per order_id

SELECT DISTINCT * FROM silver.orders;

dropDuplicates(["order_id"]) keeps one row per key, but it does not promise which one. If rows for the same order_id have different updated_at values, the survivor can be the old version. When "keep the latest" matters, use a window.

Keeping the latest record with a window function

ROW_NUMBER() numbers rows inside each partition of a window according to an ordering. Keeping row number 1 keeps exactly one row per key, chosen by your rule.

CREATE OR REPLACE TABLE silver.customers AS
SELECT * EXCEPT (rn)
FROM (
  SELECT *,
         ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY updated_at DESC) AS rn
  FROM bronze.customers_raw
)
WHERE rn = 1;

from pyspark.sql import Window, functions as F

w = Window.partitionBy("customer_id").orderBy(F.col("updated_at").desc())
latest = (bronze.withColumn("rn", F.row_number().over(w))
                .filter("rn = 1")
                .drop("rn"))

Two details are tested often. First, RANK() and DENSE_RANK() give the same rank to ties, so WHERE rank = 1 can return two rows for one key when two records share the latest timestamp; ROW_NUMBER() never ties. Second, add a tie-breaker to the ordering (for example updated_at DESC, ingest_id DESC) so the result is deterministic.

Concrete example. A retailer's CRM sends customer changes. Customer 8172 appears three times: address changes on 3 March, 9 March and 9 March (same day, two corrections, ingest ids 9001 and 9002). dropDuplicates(["customer_id"]) might keep the 3 March row. A window with ORDER BY updated_at DESC, ingest_id DESC keeps ingest 9002, the true latest.

Deduplicating streams

For streaming data, dropDuplicates also works on a streaming DataFrame, but it must remember the keys seen so far. Without a watermark (withWatermark("event_ts", "1 hour")) that state grows forever. With a watermark on an event-time column that is part of the deduplication, Spark can drop old state. Merging into a Delta table with MERGE INTO on the key is the other common pattern for idempotent loads.

2. Aggregations

An aggregation collapses many rows into one per group.

SELECT country,
       count(*)                     AS rows_in_group,
       count(discount)              AS rows_with_discount,
       count(DISTINCT customer_id)  AS buyers,
       round(avg(amount), 2)        AS avg_amount,
       sum(amount)                  AS revenue
FROM silver.orders
GROUP BY country;

(orders.groupBy("country")
       .agg(F.count("*").alias("rows_in_group"),
            F.count("discount").alias("rows_with_discount"),
            F.countDistinct("customer_id").alias("buyers"),
            F.round(F.mean("amount"), 2).alias("avg_amount"),
            F.sum("amount").alias("revenue")))

Know these behaviors:

  • count(*) counts rows. count(col) counts rows where col is not null. This difference is a favorite question.
  • avg/mean, sum, min and max ignore nulls. If every value is null, sum and avg return null, not zero.
  • Every column in the SELECT list must be either in GROUP BY or inside an aggregate function.
  • HAVING filters groups after aggregation; WHERE filters rows before it.

Approximate distinct counts

An exact count(DISTINCT ...) on a huge, high-cardinality column needs to shuffle and hold every distinct value. approx_count_distinct uses the HyperLogLog++ algorithm, memory-bounded and much faster, with a default relative standard deviation of about 5%. You can tighten it (approx_count_distinct(col, 0.01)) at a higher cost.

SELECT approx_count_distinct(user_id) AS approx_users FROM bronze.clickstream;

Use it for dashboards and profiling where a 1% to 5% error is acceptable. Use the exact version for billing or reconciliation, where it is not.

Profiling with describe and summary

df.describe() returns count, mean, stddev, min and max for numeric and string columns. df.summary() returns the same plus quartiles (25%, 50%, 75%) by default, and you can choose statistics, for example df.summary("count", "min", "max", "75%"). In notebooks, display(df.summary()) and the built-in data profile tab give the same insight. These are quick tools to spot nulls (a count lower than the table size), outliers (a max far from the 75% quartile) and wrong types.

3. Basic Spark tuning

Spark splits data into partitions, and each partition is processed by a task. Operations like joins and aggregations need a shuffle: data is redistributed across the cluster by key, which is expensive because it involves network and disk. Most basic tuning is about the number and size of partitions and about memory.

Property What it controls Default Typical symptom and action
spark.sql.shuffle.partitions Number of partitions after a shuffle in DataFrame/SQL joins and aggregations 200 (can be set to auto on Databricks) Many tiny tasks on small data: lower it. Huge tasks that spill to disk on large data: raise it, or use auto
spark.default.parallelism Default partition count for RDD operations such as parallelize and RDD shuffles Total cores in the cluster (typically) Does not affect DataFrame shuffles; use for RDD code
spark.executor.memory Heap memory per executor Set by the node type on Databricks Out-of-memory errors and spill in executors
spark.driver.memory Heap memory of the driver Set by the node type Driver out of memory after collect() or large toPandas()
spark.sql.autoBroadcastJoinThreshold Maximum size of a table that Spark broadcasts automatically in a join 10 MB (open-source default) Raise it for a medium dimension table; set to -1 to disable broadcasting

On Databricks, executor and driver memory come from the compute you select, and cluster-level properties are set in the compute's Spark configuration. Notebook-level settings use spark.conf.set("spark.sql.shuffle.partitions", "64"), which affects the current session. On serverless compute and SQL warehouses, most of these properties are managed for you and only a small allowlist can be set.

Shuffle partitions versus default parallelism

This pair is a classic trap. spark.sql.shuffle.partitions governs DataFrame and SQL shuffles (groupBy, join, orderBy). spark.default.parallelism governs RDD operations. Changing parallelism to speed up a SQL GROUP BY has no effect. A useful rule of thumb for DataFrame work is to aim for shuffle partitions of roughly 100 to 200 MB each; with 200 partitions, about 20 to 40 GB of shuffled data fits that range.

Broadcast threshold

If a join side is smaller than spark.sql.autoBroadcastJoinThreshold, Spark broadcasts it and skips the shuffle of the big side. Raising the threshold too far risks driver and executor memory pressure, because the broadcast table is first collected on the driver. Databricks also applies adaptive execution at runtime and can convert a sort-merge join to a broadcast join when the real size after filtering is small, using a separate threshold (30 MB by default for the adaptive conversion). A hint such as broadcast(df) is the most predictable choice when you know the table is small.

Adaptive query execution and auto-optimized shuffle

Adaptive query execution (AQE) is enabled by default on Databricks. It re-plans while the query runs, using real statistics: it coalesces small shuffle partitions, switches join strategies, and splits skewed partitions in shuffle joins. Setting spark.sql.shuffle.partitions to auto lets Databricks choose the partition number from the query plan and input size (auto-optimized shuffle). Because of AQE, many manual tuning steps are less necessary than in older Spark versions. The exam tends to ask which setting addresses which symptom, and not to compute values.

Worked example: diagnosing a slow nightly aggregation

Scenario. A nightly job aggregates 600 GB of order lines into daily revenue per store with a groupBy. The Spark UI shows the aggregation stage has 200 tasks, each processing about 3 GB, and many tasks spill gigabytes to disk. The job takes 55 minutes. A teammate proposes raising spark.default.parallelism to 2,000.

Step 1: identify what the stage uses. The job uses DataFrames and a SQL aggregation, so the shuffle partition count comes from spark.sql.shuffle.partitions, which is the default 200. spark.default.parallelism is for RDDs and would change nothing here.

Step 2: compute a sensible target. 600 GB divided by about 200 MB per partition suggests roughly 3,000 shuffle partitions. Tasks of 3 GB are too big, which explains the spill.

Step 3: apply one change.

spark.conf.set("spark.sql.shuffle.partitions", "3000")

Alternatively, set it to auto on supported compute and let auto-optimized shuffle decide.

Step 4: re-measure. Run again and compare duration, spill and the task time distribution in the Spark UI. Suppose the stage drops to 14 minutes with no spill. The change is confirmed. If the time had not improved, you would revert and investigate skew rather than stack more changes.

Why the other options are worse. Raising spark.default.parallelism does not affect a DataFrame shuffle. Doubling spark.executor.memory might reduce spilling, but it costs more and leaves oversized tasks. Switching to an exact count(DISTINCT) would add work. A broadcast hint is irrelevant, since there is no join.

Worked example: choosing the right deduplication

Scenario. A bank loads bronze.accounts_raw with 5 million rows. Account 55012 appears twice: balance 1,200 at 08:00 and balance 1,450 at 09:30. Another account, 55013, appears twice with identical values in every column. The silver table must hold one row per account with the most recent balance.

Step 1: classify the duplicates. Account 55013 is an exact duplicate; distinct() would remove it. Account 55012 is a key duplicate with different content; distinct() keeps both rows because they are not identical.

Step 2: reject the shortcut. dropDuplicates(["account_id"]) gives one row per account but may keep the 08:00 balance of 1,200, which is stale.

Step 3: use a window. Partition by account_id, order by updated_at DESC, keep ROW_NUMBER() = 1. Account 55012 keeps the 09:30 row with 1,450, and account 55013 keeps one of its identical rows.

Step 4: validate. Check that count(*) equals count(DISTINCT account_id) in the result. If they differ, the deduplication is incomplete.

Why the others are worse. distinct() leaves the stale rows, dropDuplicates on the key may keep the wrong version, and RANK() could keep two rows if two updates share one timestamp.

Measuring and re-measuring

Tuning without measurement is guessing. A disciplined workflow:

  1. Record a baseline: run time, Spark UI stage durations, shuffle read/write sizes, spill and the task duration spread, ideally across several runs because caches and cluster state vary.
  2. Change one setting.
  3. Re-run and compare with the same data and compute.
  4. Keep the change only if the metric improves; document it.

Also remember that performance can change when data grows, so a setting tuned for last year's volume may be wrong today. Prefer adaptive features (AQE, auto shuffle, liquid clustering) over hard-coded values when they are available.

Common pitfalls and exam tips

  • count(*) versus count(col). The second ignores nulls.
  • dropDuplicates(keys) keeps an arbitrary row. Use ROW_NUMBER() when you need the latest.
  • RANK ties. WHERE rank = 1 may return several rows per key. ROW_NUMBER returns exactly one.
  • union plus distinct is costly. It shuffles all data; deduplicate on the smallest key set you can.
  • Parallelism versus shuffle partitions. DataFrame shuffles use spark.sql.shuffle.partitions.
  • Too many or too few partitions. Thousands of tiny tasks waste scheduling time; a few giant tasks spill and run slowly.
  • Driver memory. collect() and toPandas() bring all data to the driver and can crash it.
  • Approximate means approximate. Never use approx_count_distinct where exact numbers are contractually required.

Exam tip: Match symptom to setting. Tasks too large and spilling on a DataFrame shuffle points to more shuffle partitions. A small join side that still shuffles points to a broadcast hint or a higher broadcast threshold. A driver crash after collecting points to driver memory or avoiding collect(). Every tuning answer should include re-measuring.

Mini case study

Situation. A media company ingests viewing events with Auto Loader. Because upstream retries, the same event can arrive up to three times, with identical event_id and payload but different ingest_time. Product analytics needs the number of distinct viewers per day in a dashboard (error of 3% is acceptable), and finance needs the exact count of billable events. The daily job currently takes 40 minutes and its Spark UI shows 200 shuffle tasks of about 2.5 GB each with heavy spill.

Task. Propose the deduplication, aggregation and tuning approach.

Solution.

  1. Deduplicate on the key. Duplicates share event_id, and the contents are identical, so dropDuplicates(["event_id"]) is sufficient; which copy survives does not matter. If the payload could differ, a window ordered by ingest_time DESC would be needed.
  2. Aggregate. Finance uses count(*) on the deduplicated table for the exact billable count. The dashboard uses approx_count_distinct(viewer_id) because a 3% error is acceptable and the computation is faster.
  3. Tune. The tasks are far too large, so increase spark.sql.shuffle.partitions (or use auto) and leave spark.default.parallelism alone, since there is no RDD work.
  4. Verify. Compare duration and spill with the baseline in the Spark UI, keep the setting only if it helps.

Key terms

  • Shuffle: redistribution of data across partitions by key, needed for joins and aggregations.
  • Partition: a chunk of data processed by one task.
  • dropDuplicates: removes rows with the same values in the chosen columns, keeping an arbitrary one.
  • ROW_NUMBER(): window function giving a unique sequence number per ordered partition.
  • approx_count_distinct: fast, memory-bounded estimate of distinct values (HyperLogLog++).
  • describe / summary: quick descriptive statistics of a DataFrame.
  • AQE: adaptive query execution, re-optimization at runtime from actual statistics.
  • Spill: writing intermediate data to disk when memory is insufficient.

Summary checklist

  • I can remove exact and key-based duplicates and keep the latest row with ROW_NUMBER().
  • I know count(*) counts rows and count(col) ignores nulls.
  • I can use approx_count_distinct, describe and summary, and know when approximate is acceptable.
  • I can match spark.sql.shuffle.partitions, spark.default.parallelism, executor and driver memory and the broadcast threshold to their symptoms.
  • I know AQE and auto shuffle reduce manual tuning.
  • I measure a baseline, change one setting, and re-measure.

Official sources