PySpark Deduplication: distinct() vs dropDuplicates()

1.distinct(): removing rows that match in every column

Bronze data often contains exact copies of the same record, for example when a source system sends the same event twice. The simplest tool for this is distinct(). It takes no arguments and returns a new DataFrame that contains only the distinct rows. Two rows count as duplicates only when every column matches. Like other DataFrame transformations, it leaves the original DataFrame unchanged and gives you a new one.

df = spark.createDataFrame(
    [(14, "Tom"), (23, "Alice"), (23, "Alice")], ["age", "name"])
df.distinct().show()
# +---+-----+
# |age| name|
# +---+-----+
# | 14|  Tom|
# | 23|Alice|
# +---+-----+

df.distinct().count()
# 2

Chaining count() after distinct() is the usual way to find out how many unique records a dataset really holds. Adding columns doesn't change the rule. In a second documented example, a three-column DataFrame with the rows (14, Tom, M), (23, Alice, F), (23, Alice, F) and (14, Tom, M) also reduces to two rows, because each pair matches in all three columns. distinct() has no parameter, so you can't tell it to compare only some of the columns. If two records share a business key but differ in a timestamp or some other column, distinct() keeps both.

Sources1

2.dropDuplicates(subset): deduplicating on key columns

dropDuplicates() covers both cases. Called with no arguments it compares all columns, just like distinct(). Its optional subset parameter takes a list of column names, and only those columns are used to decide whether two rows are duplicates. This is how you deduplicate on a business key while other columns differ.

from pyspark.sql import Row
df = spark.createDataFrame([
    Row(name='Alice', age=5, height=80),
    Row(name='Alice', age=5, height=80),
    Row(name='Alice', age=10, height=80)
])

df.dropDuplicates().show()
# +-----+---+------+
# | name|age|height|
# +-----+---+------+
# |Alice|  5|    80|
# |Alice| 10|    80|
# +-----+---+------+

df.dropDuplicates(['name', 'height']).show()
# +-----+---+------+
# | name|age|height|
# +-----+---+------+
# |Alice|  5|    80|
# +-----+---+------+

The row with age 10 is gone, even though its age differs. With a subset, the columns outside the subset don't stop two rows from being duplicates, but the row that survives still carries a value for each of them. In this example the kept row has age 5. The reference page doesn't say which row of a duplicate group is kept, so don't build logic on the assumption that a particular one, such as the first or the latest, will win.

df. ? (['name', 'height']).show()

Sources2

3.dropDuplicates on streaming DataFrames: state and watermarks

The same method behaves differently depending on the kind of DataFrame. On a static batch DataFrame, all the rows are available at once, so dropDuplicates just removes the duplicates and finishes. A streaming DataFrame never has all of its data at once. A duplicate of a record seen in an earlier trigger could arrive in a later one. To catch it, Spark must remember what it has already seen.

The documentation answers this directly. For a streaming DataFrame, dropDuplicates keeps all data across triggers as intermediate state, so with no limit that state keeps growing. The control for this is withWatermark. A watermark sets how late duplicate data can be, and the system limits the state to match. There is a cost: data older than the watermark is dropped, so a very late duplicate can't slip through as a new row.

Sources2

Exam traps

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

1. 1.distinct() can be given a list of key columns so that it deduplicates on those columns only.Why is that wrong?

   distinct() takes no parameters and always compares whole rows. To deduplicate on chosen columns, use dropDuplicates with a subset.

   Covered in distinct(): removing rows that match in every column

2. 2.dropDuplicates works the same way on a streaming DataFrame as on a batch DataFrame, at no extra cost.Why is that wrong?

   On a streaming DataFrame it keeps data across triggers as intermediate state. You need withWatermark to bound that state, and data older than the watermark is dropped.

   Covered in dropDuplicates on streaming DataFrames: state and watermarks