How to drop duplicates in Delta Table?

Yes you can delete duplicates directly from delta table. Using merge command. Below command will retain only latest records and rest redundant data is deleted.

MERGE into [deltatable] as target
USING ( select *, ROW_NUMBER() OVER (Partition By [primary keys] Order By [date] desc) as rn  from [deltatable]) t1 where rn> 1 ) as source
ON [merge primary keys and date column between source and target]
WHEN MATCHED THEN DELETE

There are other multiple ways as well but they are time consuming.

If you have a primary key (like UUID), and the duplicates are based on a certain combination of columns (e.g., Col1, Col2, Col3), then you can use the ROW_NUMBER() approach to get a list of UUIDs of the duplicate rows to delete. As a side note, Delta tables don't have ROWID or sequences to auto-generate primary keys at this time.

If your duplicates are based on a certain composite key (e.g., Col2, Col4, Col7), the ROW_NUMBER() trick to get rid of the duplicates will not work: it will delete all copies of the row. For this scenario, you can get rid of the duplicates in a Delta table using a combination of Delta MERGE and Delta Time Travel (versioning) features. Here are the steps to eliminate the full duplicates (the rows where all the corresponding fields have identical values):

1. Get a dataframe with the distinct rows that have duplicates in the Delta table. ROW_NUMBER() function will help you here.
2. Use MERGE operation and WHEN MATCHED DELETE to remove these rows. Notice that if you have 4 copies of the same row, it will delete all 4 copies. We will re-add non-duplicates in Step 5.
3. Use DESCRIBE HISTORY command to get the version number of the Delta table before the current. That version still has all the duplicates, while the current version doesn't.
4. Repeat Step 1, but this time use VERSION AS OF option to get the dataframe with the distinct rows we deleted in Step 2.
5. Use MEREGE operation and WHEN NOT MATCHED INSERT ALL to add the distinct rows that used to have duplicates.

Here is a Python example that I tested on Azure Databricks with a Delta table stored in ADLS.

from delta.tables import *

# Step 1
dfTemp = (
  spark.sql("SELECT *, ROW_NUMBER() OVER (PARTITION BY Col2, Col4, Col7 ORDER BY Col9 DESC) rn FROM delta.`abfss://adls-account@adls-store.dfs.core.windows.net/my-path/my-delta-table`")
           ).filter(f.col('rn') > 1).drop('rn').distinct()

# Step 2
deltaTemp = DeltaTable.forPath(spark, "abfss://adls-account@adls-store.dfs.core.windows.net/my-path/my-delta-table")

deltaTemp.alias("main").merge(
    dfTemp.alias("nodups"),
    "main.Col2 = nodups.Col2 AND main.Col4 = nodups.Col4 AND main.Col7 = nodups.Col7").whenMatchedDelete().execute()

Now, get the version number of the Delta table before the merge we performed in Step 2 above. I ran this command as SQL in my Azure Databricks notebook:

%sql

-- Step 3
DESCRIBE HISTORY delta.`abfss://adls-account@adls-store.dfs.core.windows.net/my-path/my-delta-table`

Let's say, my current Delta table version is 50, and my previous version (with the duplicates) was 49. You can now execute the remaining steps below:

# Step 4
dfTemp = (
  spark.sql("SELECT *, ROW_NUMBER() OVER (PARTITION BY Col2, Col4, Col7 ORDER BY Col9 DESC) rn FROM delta.`abfss://adls-account@adls-store.dfs.core.windows.net/my-path/my-delta-table` VERSION AS OF 49")
           ).filter(f.col('rn') > 1).drop('rn').distinct()

# Step 5
deltaTemp = DeltaTable.forPath(spark, "abfss://adls-account@adls-store.dfs.core.windows.net/my-path/my-delta-table")

deltaTemp.alias("main").merge(
    dfTemp.alias("nodups"),
    "main.Col2 = nodups.Col2 AND main.Col4 = nodups.Col4 AND main.Col7 = nodups.Col7").whenNotMatchedInsertAll().execute()

When you have to deal with partial duplicates (the rows where the primary keys are the same, but some other corresponding fields have different values), you will probably end up coding the logic for "marking" which one of the duplicate rows to keep. The logic will fully depend on your particular use case.

we assume the primary key is columns a,b,c , and the timestamp column is d, and we gonna choose the latest version of the data

MERGE into delta.`/mnt/lei/dupTab` as target
USING (
      with t as(
      select a,b,c, d,ROW_NUMBER() OVER (Partition By a,b,c Order By d desc) as rn  from delta.`/mnt/lei/dupTab`
      )
      select * from t where rn > 1
 
) 
as source
ON source.a=target.a and source.b=target.b and source.c=target.c and source.d=target.d
WHEN MATCHED THEN DELETE