Definition
- distinct() – removes rows where every column value matches another row; deduplicates on the entire row.
- dropDuplicates(subset) – removes rows where the specified columns match, regardless of other columns; keeps one row per distinct combination.
- No-argument dropDuplicates() – behaves identically to distinct().
- Business key – in banking, the dedup key is almost always a subset like transaction id, not the whole row.
- Survivor selection – dropDuplicates keeps an arbitrary row; use a window function for deterministic survivor choice.
At the National Bank of India, imagine a stack of transaction forms. distinct() throws away a form only if another form is identical in every single field. dropDuplicates on transaction id throws away a form if another has the same id, even if the timestamp or channel differs. If the upstream system re-sent the same transaction with a different received-time, distinct would keep both; dropDuplicates on the id correctly treats them as one.
Understanding the Concepts
What is the difference between distinct and dropDuplicates?
distinct() deduplicates on all columns. dropDuplicates(subset) deduplicates on a chosen set of columns. This difference determines whether near-duplicates with differing metadata are collapsed.
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
spark = SparkSession.builder.appName("DedupDemo").getOrCreate()
# Transaction data with duplicate deliveries at the National Bank of India
txns = spark.createDataFrame([
("TXN001", "SAV2001", 50000.00, "2024-01-15 10:00", "API"),
("TXN001", "SAV2001", 50000.00, "2024-01-15 10:05", "RETRY"), # re-delivery
("TXN002", "CUR3001", 75000.00, "2024-01-15 11:00", "API"),
("TXN002", "CUR3001", 75000.00, "2024-01-15 11:00", "API"), # exact copy
], ["txn_id", "account_id", "amount", "received_at", "channel"])
How does distinct() work?
distinct() considers two rows duplicates only when every column matches. Rows with any difference are kept.
distinct_result = txns.distinct()
distinct_result.show()
# TXN001 appears TWICE because received_at and channel differ
# TXN002 appears ONCE because the exact duplicate was removed
+------+----------+--------+----------------+-------+
|txn_id|account_id| amount| received_at|channel|
+------+----------+--------+----------------+-------+
|TXN001| SAV2001|50000.00|2024-01-15 10:00| API|
|TXN001| SAV2001|50000.00|2024-01-15 10:05| RETRY| <- kept!
|TXN002| CUR3001|75000.00|2024-01-15 11:00| API|
+------+----------+--------+----------------+-------+
How does dropDuplicates(subset) work?
dropDuplicates with a subset deduplicates on just those columns. Any difference in other columns is ignored.
# Deduplicate on business key (txn_id)
deduped = txns.dropDuplicates(["txn_id"])
deduped.show()
# TXN001 appears ONCE: the re-delivery is correctly removed
+------+----------+--------+----------------+-------+
|txn_id|account_id| amount| received_at|channel|
+------+----------+--------+----------------+-------+
|TXN001| SAV2001|50000.00|2024-01-15 10:00| API|
|TXN002| CUR3001|75000.00|2024-01-15 11:00| API|
+------+----------+--------+----------------+-------+
Which row does dropDuplicates keep?
An important subtlety: dropDuplicates keeps an arbitrary surviving row, not necessarily the first or latest. Which row survives depends on partitioning and task execution order.
# WARNING: dropDuplicates does NOT guarantee keeping the latest
# It keeps an arbitrary row per key combination
# If you need the latest received_at for each txn_id:
# Use a window function for deterministic selection
from pyspark.sql.functions import row_number
from pyspark.sql.window import Window
w = Window.partitionBy("txn_id").orderBy(col("received_at").desc())
deduped_latest = txns \
.withColumn("rn", row_number().over(w)) \
.filter(col("rn") == 1) \
.drop("rn")
deduped_latest.show()
# Guaranteed to keep the latest received_at for each txn_id
When do you use each?
| Scenario | Use | Why |
|---|---|---|
| Remove exact duplicate rows | distinct() | Whole-row comparison |
| Remove duplicate deliveries by business key | dropDuplicates(["txn_id"]) | Business key defines uniqueness |
| Keep the latest delivery per key | Window + row_number | Deterministic survivor selection |
| After a union that may produce copies | distinct() | Exact copies from both sides |
| Dedup on composite key | dropDuplicates(["account_id", "txn_date"]) | Multi-column business key |
# After union: use distinct for exact copies
df1 = spark.read.parquet("/data/batch1")
df2 = spark.read.parquet("/data/batch2")
combined = df1.union(df2).distinct() # Remove exact duplicates
# Business dedup: use dropDuplicates on the key
deduped = combined.dropDuplicates(["txn_id"])
What about performance?
Both operations require a shuffle because Spark must bring matching rows together. Performance is comparable; the choice is about correctness.
# Both are wide operations with a shuffle
# distinct: shuffles on ALL columns (hash of entire row)
# dropDuplicates(subset): shuffles on the subset columns only
# dropDuplicates on a small subset may shuffle less data
# because the hash is computed on fewer columns
# But the dominant cost is the shuffle itself, which is similar
What about Structured Streaming?
In Structured Streaming, dropDuplicates is stateful: it remembers seen keys across micro-batches. Pair it with a watermark to bound the state.
# Streaming dedup with watermark
stream = spark.readStream \
.format("kafka") \
.option("subscribe", "transactions") \
.load()
deduped_stream = stream \
.withWatermark("event_time", "1 hour") \
.dropDuplicates(["txn_id", "event_time"])
# State only remembers keys within the watermark window
# Without watermark, state grows unbounded!
What are common mistakes?
The two mistakes a senior engineer watches for are choosing the wrong dedup method and assuming the survivor.
# MISTAKE 1: using distinct when you need a subset key
# Near-duplicates with different metadata survive distinct()
bad = txns.distinct() # TXN001 appears twice!
# CORRECT: use dropDuplicates on the business key
good = txns.dropDuplicates(["txn_id"]) # TXN001 appears once
# MISTAKE 2: assuming dropDuplicates keeps the latest
bad_assumption = txns.dropDuplicates(["txn_id"])
# The survivor may be the API or RETRY delivery arbitrarily!
# CORRECT: use window function for deterministic selection
w = Window.partitionBy("txn_id").orderBy(col("received_at").desc())
good_latest = txns.withColumn("rn", row_number().over(w)) \
.filter(col("rn") == 1).drop("rn")
Visual Overview
distinct() – whole-row comparison ┌────────────────────────────────────────────┐ │ TXN001 SAV2001 50000 10:00 API │ kept │ TXN001 SAV2001 50000 10:05 RETRY │ kept (different!) │ TXN002 CUR3001 75000 11:00 API │ kept │ TXN002 CUR3001 75000 11:00 API │ removed (exact) └────────────────────────────────────────────┘ Result: 3 rows dropDuplicates(["txn_id"]) – subset comparison ┌────────────────────────────────────────────┐ │ TXN001 SAV2001 50000 10:00 API │ kept │ TXN001 SAV2001 50000 10:05 RETRY │ removed (same id) │ TXN002 CUR3001 75000 11:00 API │ kept │ TXN002 CUR3001 75000 11:00 API │ removed (same id) └────────────────────────────────────────────┘ Result: 2 rows
Key Takeaway
distinct and dropDuplicates both remove duplicates, but they define "duplicate" differently. distinct considers two rows duplicates only when every column matches, deduplicating on the whole row. dropDuplicates with a subset considers rows duplicates when just those columns match, ignoring the others. This distinction matters because duplicate deliveries of the same transaction typically differ in metadata like the received timestamp or channel, so distinct keeps both while dropDuplicates on the transaction id correctly collapses them.
The subtle trap is that dropDuplicates keeps an arbitrary surviving row, not necessarily the latest. When the surviving row matters, the correct pattern is a window function partitioned by the key and ordered by a tiebreaker, filtering where row_number equals one. This both deduplicates and guarantees the intended survivor.
At the National Bank of India, ingestion deduplicated transactions on the transaction id because the upstream system sometimes re-sent the same transaction with a different received-timestamp. Using dropDuplicates on the id was correct where distinct would have kept both copies. Where the latest copy was needed for the ledger, a window function with row_number was used.
The two mistakes to avoid are choosing distinct when you need a subset key, which leaves near-duplicates behind, and using dropDuplicates while assuming it keeps the latest, which silently keeps an arbitrary row.
Remember one sentence: distinct deduplicates on all columns, dropDuplicates deduplicates on a chosen key, and when the surviving row matters use a window function with row_number.
---