Deduplication, Aggregation and Basic Spark Tuning in Databricks (Data Engineer Associate)

Deduplication starts with one decision: which columns define a duplicate. distinct() and dropDuplicates() with no arguments treat two rows as duplicates only when every column is equal, which is right for exact copies such as a bronze load that ran twice. dropDuplicates(["col1", "col2"]) compares only the listed columns and keeps one full row per combination.

Two mistakes come from choosing the wrong column set. A subset that is too narrow (for example order_id alone when an order has several lines, or when rows sharing an id are legitimate corrections) merges rows that are genuinely different. A subset that is too wide (including load_ts or ingest_file) keeps copies that differ only in metadata. The right subset is the natural key of the entity. dropna() is not a deduplication tool: it removes rows with nulls.

In SQL, SELECT DISTINCT * is the whole-row equivalent. GROUP BY on a key with SELECT * is not valid deduplication, because every non-grouped column must be aggregated.

When the requirement is a specific row per key, such as the current price for each product, use a window function, not dropDuplicates. dropDuplicates(subset) keeps one row per key but makes no promise about which one; sorting the DataFrame first does not help, because the deduplication happens after a shuffle that does not preserve that order.

from pyspark.sql import Window
from pyspark.sql import functions as F

w = (Window.partitionBy("product_id")
     .orderBy(F.col("effective_at").desc(),
              F.col("price_version").desc()))

latest = (df.withColumn("rn", F.row_number().over(w))
            .filter("rn = 1")
            .drop("rn"))

The pattern is: partition by the key, order by the recency column descending, number the rows with row_number(), keep row 1. Alternatives fall short in specific ways:

• groupBy(key).agg(max("effective_at")) returns the key and the timestamp only; the other columns are lost unless you join back.
• dropDuplicates(["product_id", "effective_at"]) keeps one row per key and timestamp, so a product with three price changes still has three rows.
• rank() and dense_rank() give tied rows the same number, so filtering on 1 keeps every tied row and the key appears twice.

Ties and repeatable results

If two rows in a partition share the same ordering value, row_number() numbers them in whatever order they arrive, which can change between runs. To make the result deterministic, add a unique tie-breaker after the recency column, such as a version number. The rule is that the column carrying the business meaning of "latest" comes first in the ORDER BY and the unique column comes last, where it only decides between rows that are otherwise tied.

The same pattern in Databricks SQL: QUALIFY

Databricks SQL supports QUALIFY, a clause that filters on the result of window functions, so the keep-the-latest pattern fits in one SELECT without a subquery or CTE:

SELECT *
FROM product_prices
QUALIFY ROW_NUMBER() OVER (
  PARTITION BY product_id
  ORDER BY effective_at DESC, price_version DESC) = 1

The reason QUALIFY exists is evaluation order. A query evaluates WHERE, then GROUP BY, then HAVING, and only then computes window functions. A window function therefore cannot appear in WHERE or HAVING; QUALIFY runs after the windows are computed. The same tie rules apply as in PySpark: use ROW_NUMBER() (not RANK()) when the key must appear exactly once, and add a unique tie-breaker last in the ORDER BY when ties are possible.

On a Structured Streaming DataFrame, dropDuplicates works across micro-batches by remembering every key it has seen in the state store. Without a watermark that state grows forever. A watermark (withWatermark("event_ts", "10 minutes")) lets Spark discard old state, but which deduplication call you use decides whether it can.

So: if duplicates are exact copies with the same timestamp, include the event-time column in the dropDuplicates key. If retries carry a new timestamp (a producer resending an event), use dropDuplicatesWithinWatermark with a watermark delay longer than the maximum gap between duplicates. Deduplicating inside foreachBatch only removes duplicates within one micro-batch.

Spark's counting functions differ in how they treat nulls, and that is the source of most "the numbers don't match" puzzles.

Examples: 500 rows where 60 have a null phone and the other 440 phones include 400 distinct numbers give count("*") = 500, count("phone") = 440, countDistinct("phone") = 400. A status column with three values plus some nulls gives 4 from select("status").distinct().count() and 3 from countDistinct("status"). When a null must count as its own value (a web session with no referrer is still a distinct page/referrer pair), use select(...).distinct().count().

To count missing values, subtract: F.count("*") - F.count("amount"). F.count(F.isnull("amount")) looks right but counts every row, because isnull returns true or false, never null, and count only skips nulls. (Summing a 0/1 flag such as F.sum(F.col("amount").isNull().cast("int")) also works.)

approx_count_distinct estimates the number of distinct values with the HyperLogLog++ algorithm. An exact distinct count has to track every distinct value; HLL++ keeps a small fixed-size sketch per group, so it is much cheaper on billions of rows when a few percent of error is acceptable.

• The rsd argument is the maximum relative standard deviation, as a fraction. The default is 0.05 (about 5%). Lower values (for example rsd=0.01) give a tighter estimate and a bigger sketch; higher values loosen it.
• The Spark documentation notes that for rsd below 0.01 an exact count_distinct is usually more efficient.
• Accuracy depends only on rsd. Shuffle partitions, caching or cluster size change speed, not the error.

Distinct counts do not add up

Distinct counts are not additive. A user active on five days appears in five daily counts, so summing daily unique users overstates weekly unique users; taking the maximum understates it; and counting the distinct daily totals measures nothing useful. This is true for exact and approximate counts alike. To roll up without rescanning raw events, store a sketch instead of a number. Databricks SQL and PySpark provide:

A weekly figure from daily sketches is hll_sketch_estimate(hll_union_agg(daily_sketch)), grouped by week. Any period can be rolled up the same way.

describe() and summary() both return a quick statistical profile as a DataFrame with a summary column holding the statistic name and one column per profiled data column. They differ in what the arguments mean and what they compute.

So df.summary("count", "min", "max", "50%") returns exactly those four statistics; to limit columns, select them first: df.select("amount", "qty").summary(...). Because statistics are rows of the output, you filter the summary column rather than selecting statistic columns. printSchema() and dtypes describe types, not values. In a notebook, display() output also offers a data profile, but describe()/summary() are the DataFrame API answers.

avg (alias mean) and the other aggregate functions skip nulls: avg("rating") divides the sum of non-null ratings by the number of non-null ratings, which is the same as sum("rating") / count("rating"). Whether that is right depends on what a null means in the business rule.

Be careful with dropna() without a subset: it removes rows with a null in any column, so an unrelated nullable column (a missing comment) can silently remove valid values from the average. Use dropna(subset=["rating"]) or rely on the function's null skipping.

One groupBy, many aggregates

groupBy() returns grouped data; agg() takes any number of Column expressions and computes them all in one pass:

(df.groupBy("region")
   .agg(F.count("*").alias("orders"),
        F.avg("amount").alias("avg_amount"),
        F.max("amount").alias("max_amount"),
        F.approx_count_distinct("customer_id")
          .alias("customers")))

Avoid the dictionary form for several functions on one column: {"amount": "avg", "amount": "max"} is a Python dict, which silently keeps only the last value for a repeated key. Aggregating before grouping (df.agg(...).groupBy(...)) collapses the DataFrame to one row first; adding the measured column to groupBy changes the grain; and separate aggregations joined back, or windows followed by deduplication, do the same work several times.

Four groups of settings cover the basic tuning the exam guide names. Each controls something different, and a change only helps if it targets what the Spark UI shows is wrong.

Shuffle partitions

Each shuffle partition becomes one task. Too many partitions for small data means hundreds of tasks processing a few kilobytes each, so scheduling overhead dominates: lower the value. Too few for big data means huge tasks: 1.5 TB over the default 200 partitions puts about 7.5 GB in each task, which cannot fit in executor memory and spills to disk. Raise the value so each partition holds roughly 100–200 MB (several thousand partitions for terabytes). Spill with even key distribution is a partition-size problem; spill concentrated in a few tasks is skew, which is a different fix.

Adaptive query execution (AQE), on by default, coalesces small adjacent post-shuffle partitions and splits skewed outlier partitions in joins. It does not split partitions that are all evenly too large, so for big aggregations the starting value of spark.sql.shuffle.partitions still matters: set it high enough and let AQE merge any that turn out small. Databricks also accepts spark.sql.shuffle.partitions=auto (auto-optimized shuffle), which lets the platform choose the number from the data.

spark.default.parallelism does not touch DataFrames

Raising spark.default.parallelism does not change the task count of a DataFrame groupBy or join; those read spark.sql.shuffle.partitions. A job that mixes an RDD reduceByKey and a DataFrame groupBy needs both settings to control both stages. File-scan settings such as spark.sql.files.maxPartitionBytes size input partitions, not shuffle partitions.

Driver and executor memory

Executors do the distributed work; the driver plans it and receives anything returned to the notebook. collect() and toPandas() copy every row into the driver, so a multi-gigabyte toPandas() fails with a driver out-of-memory error while executors sit idle; that is governed by spark.driver.memory (on Databricks, by the driver node type). Executor spill and executor OOM in shuffles relate to spark.executor.memory and to partition size. Because memory sizes the JVM processes, spark.conf.set("spark.executor.memory", ...) fails in Spark 3 with an error that the config cannot be modified; set it in the cluster's Spark config or pick a larger node type, then restart. When a restart is not allowed, the session-level lever is more shuffle partitions, so each task holds less data. On serverless compute, Databricks manages memory and most Spark settings itself and only a small set of session configs can be changed.

When one side of a join is estimated to be smaller than spark.sql.autoBroadcastJoinThreshold (10 MB by default), Spark plans a broadcast hash join: it collects that side on the driver and sends a copy to every executor, so the large side is not shuffled. Otherwise it typically plans a sort-merge join, which shuffles both sides. (Join types and the broadcast() hint as a join technique are covered with joins in the cleaning, joining and reshaping lesson; here the focus is the threshold as a tuning parameter.)

• Raise the threshold above a dimension's size (for example to 32 MB for a 25 MB table) to get automatic broadcasts with no code change; it can be set in a job cluster's Spark config.
• Too high is dangerous: if the threshold is raised to several gigabytes, a multi-gigabyte fact table qualifies, the driver collects it and runs out of memory. Executor memory and shuffle partitions do not fix that. Bring the threshold back near the default and use explicit broadcast() hints for the small tables that must be broadcast.
• -1 disables automatic broadcasts (hints still apply). spark.sql.broadcastTimeout only limits how long a broadcast may take, and spark.sql.join.preferSortMergeJoin (true by default) chooses between shuffle-based join types; neither turns broadcasting off.

AQE can change the join at runtime

The static planner uses estimated sizes, and a selective filter often does not shrink the estimate much, so a filtered large dimension may be planned as a sort-merge join. With AQE on, Spark measures the real size after the filtering stage and, if it is small enough, switches to a broadcast hash join. On Databricks this runtime decision uses its own setting, spark.databricks.adaptive.autoBroadcastJoinThreshold (30 MB by default). That is why a benchmark that needs a guaranteed sort-merge join turns AQE off as well as setting the static threshold to -1.

A tuning change is only finished when you have re-measured it fairly. The method:

1. Change one setting at a time, so you know which change made the difference.
2. Check the plan before running with explain(): BroadcastHashJoin confirms a broadcast, SortMergeJoin means it was not chosen. Reading the config value back only proves the setting is set, not that the planner used it. With AQE on, the output is wrapped in AdaptiveSparkPlan isFinalPlan=false: it is the initial plan, and the plan that actually ran appears in the SQL / DataFrame tab of the Spark UI after execution.
3. Run the whole transformation. count() lets Spark prune columns and skip work, and show() computes only a few rows, so neither reflects the real job. A write with df.write.format("noop").mode("overwrite").save() executes every transformation without the cost of writing real output.
4. Compare like with like. Repeat runs can be faster because of disk caching, so do not keep only the fastest of several timings.
5. Read the stage metrics. In the Spark UI, the stage details page shows per-task duration, shuffle read and write, spill (memory) and spill (disk), with percentiles. Compare the affected stage before and after. The Environment tab shows settings, the Storage tab shows cached data and the Executors tab shows resources; none of them proves the spill is gone.