dropDuplicates

What it does

Step by step

Which row survives?

dropDuplicates(["user_id"]) keeps the first row it meets for each key, and "first" depends on partitioning and task timing. On a different day, with different file sizes, it can keep a different row. It is fine for exact duplicates; it is wrong for "keep the latest".

Keep the latest row per key

Rank the rows of each key by time with a window, and keep rank 1:

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

w = Window.partitionBy("user_id").orderBy(F.col("updated_at").desc())
result = (events
    .withColumn("row_num", F.row_number().over(w))
    .filter(F.col("row_num") == 1)
    .drop("row_num"))

SELECT user_id, email, updated_at
FROM (
  SELECT *, ROW_NUMBER() OVER (
    PARTITION BY user_id ORDER BY updated_at DESC) AS rn
  FROM events)
WHERE rn = 1

Some engines, such as Databricks SQL and Snowflake, accept QUALIFY rn = 1 to skip the subquery. The subquery form works everywhere, including open-source Spark SQL. If two rows share the latest timestamp, add a tiebreaker to the orderBy.

Under the hood

Both distinct and dropDuplicates are aggregations: Spark groups by the chosen columns and keeps one row, with the same partial aggregation before the shuffle as a groupBy. In the plan they appear as HashAggregate nodes around an Exchange.

In Structured Streaming

On a stream, dropDuplicates has to remember every key it has seen, forever, unless you bound the state with a watermark: withWatermark("event_time", "1 hour").dropDuplicates(["event_id", "event_time"]). Spark 3.5 added dropDuplicatesWithinWatermark, which deduplicates on the key alone within the watermark delay.

Common mistakes