Consider the following data frame:
from pyspark.sql import SparkSession, Window
from pyspark.sql.functions import row_number
import pandas as pd
import numpy as np
spark = SparkSession.builder.getOrCreate()
pdf = pd.DataFrame(np.random.random_integers(0, 10, size=[64, 2]), columns=['id', 'key'])
pdf['value'] = pdf['key'].apply(lambda v: 'value--%d' % v)
df = spark.createDataFrame(pdf)
It is well documented that drop duplicates will result in non-deterministic behaviour here:
df.dropDuplicates(subset=['id']).orderBy('id').show()
+---+---+--------+
| id|key| value|
+---+---+--------+
| 0| 3|value--3|
...
df.dropDuplicates(subset=['id']).orderBy('id').show()
+---+---+--------+
| id|key| value|
+---+---+--------+
| 0| 0|value--0| <--- Picked some random row
...
The reason for this is that when partitioning over subset no sort order is defined.
To remove this non-deterministic behaviour, you must use a window function:
sort_window = Window.partitionBy(['id']).orderBy(['key']) <-- Explicitly choose sort order in window
df.withColumn('rank', row_number().over(sort_window)).filter('rank == 1').drop('rank').orderBy('id').show()
+---+---+--------+
| id|key| value|
+---+---+--------+
| 0| 0|value--0| <--- Always picks the same row
...
If you wish to select distinct values from a specific column subset, the distinct function also exists, eg. df.select('id','key').distinct().
However, the spark developers continue to close issues relating to this as 'expected behaviour', and have faithfully added it to the spark 3 api.
So... here's my actual question:
It seems fantastically broken to have this broken API function, that does the wrong thing in every circumstance I can imagine. For selecting subsets distinct is the correct method to use, and in
all other circumstances, the use of dropDuplicates results in undefined non-deterministic behaviour,
which is highly undesirable in data processing workloads.
Am I missing something?
In what circumstances is it ever useful to use dropDuplicates?