Home / Tutorials / Remove duplicates

PYSPARK DATA CLEANING

Remove duplicate rows by transaction ID

Example tested with PySpark 4.2.0

A duplicated transaction can double your revenue totals. Deduplicate on the business key, not every column in the row.

Create a DataFrame with a duplicate

sales = spark.createDataFrame(
    [
        (101, "Alice", 250.50),
        (102, "Bob", 120.00),
        (102, "Bob", 120.00),
    ],
    ["transaction_id", "customer_name", "net_amount"],
)

Deduplicate on the transaction ID

clean_sales = sales.dropDuplicates(["transaction_id"])

print("before:", sales.count())
print("after:", clean_sales.count())

before: 3
after: 2

Passing ["transaction_id"] tells Spark that two rows represent the same transaction when that column matches. Calling dropDuplicates() without a subset only removes rows that match across every column.

Do not use it to choose the latest record

If duplicate IDs can contain different values, dropDuplicates() does not promise which row it will retain. Use a window ordered by a timestamp when the latest record must win.

from pyspark.sql import functions as F, Window

latest_first = Window.partitionBy("transaction_id").orderBy(
    F.col("updated_at").desc()
)

latest = (
    sales.withColumn("row_number", F.row_number().over(latest_first))
    .filter(F.col("row_number") == 1)
    .drop("row_number")
)