Keep only duplicates from a DataFrame regarding some field

by

One way to do this is by using a pyspark.sql.Window to add a column that counts the number of duplicates for each row’s ("ID", "ID2", "Number") combination. Then select only the rows where the number of duplicate is greater than 1.

import pyspark.sql.functions as f
from pyspark.sql import Window

w = Window.partitionBy('ID', 'ID2', 'Number')
df.select('*', f.count('ID').over(w).alias('dupeCount'))\
    .where('dupeCount > 1')\
    .drop('dupeCount')\
    .show()
#+---+---+------+----+------------+------------+
#| ID|ID2|Number|Name|Opening_Hour|Closing_Hour|
#+---+---+------+----+------------+------------+
#|ALT|QWA|     2|null|    08:54:00|    23:25:00|
#|ALT|QWA|     2|null|    08:53:00|    23:24:00|
#|ALT|QWA|     6|null|    08:59:00|    23:30:00|
#|ALT|QWA|     6|null|    08:55:00|    23:26:00|
#+---+---+------+----+------------+------------+

I used pyspark.sql.functions.count() to count the number of items in each group. This returns a DataFrame containing all of the duplicates (the second output you showed).

If you wanted to get only one row per ("ID", "ID2", "Number") combination, you could do using another Window to order the rows.

For example, below I add another column for the row_number and select only the rows where the duplicate count is greater than 1 and the row number is equal to 1. This guarantees one row per grouping.

w2 = Window.partitionBy('ID', 'ID2', 'Number').orderBy('ID', 'ID2', 'Number')
df.select(
        '*',
        f.count('ID').over(w).alias('dupeCount'),
        f.row_number().over(w2).alias('rowNum')
    )\
    .where('(dupeCount > 1) AND (rowNum = 1)')\
    .drop('dupeCount', 'rowNum')\
    .show()
#+---+---+------+----+------------+------------+
#| ID|ID2|Number|Name|Opening_Hour|Closing_Hour|
#+---+---+------+----+------------+------------+
#|ALT|QWA|     2|null|    08:54:00|    23:25:00|
#|ALT|QWA|     6|null|    08:59:00|    23:30:00|
#+---+---+------+----+------------+------------+

More Related Contents:

  1. Finding duplicates from large data set using Apache Spark
  2. How to melt Spark DataFrame?
  3. Using a column value as a parameter to a spark DataFrame function
  4. Find maximum row per group in Spark DataFrame
  5. Unpivot in spark-sql/pyspark
  6. Split Spark Dataframe string column into multiple columns
  7. Avoid performance impact of a single partition mode in Spark window functions
  8. How to access element of a VectorUDT column in a Spark DataFrame?
  9. Spark Dataframe validating column names for parquet writes
  10. How to check if spark dataframe is empty?
  11. How to split a list to multiple columns in Pyspark?
  12. pyspark dataframe filter or include based on list
  13. How to save/insert each DStream into a permanent table
  14. Pyspark : forward fill with last observation for a DataFrame
  15. Spark load data and add filename as dataframe column
  16. PySpark: how to resample frequencies
  17. How to loop through each row of dataFrame in pyspark
  18. Fill in null with previously known good value with pyspark
  19. Why does Spark think this is a cross / Cartesian join
  20. PySpark – get row number for each row in a group
  21. How to find count of Null and Nan values for each column in a PySpark dataframe efficiently?
  22. Rename more than one column using withColumnRenamed
  23. pyspark: count distinct over a window
  24. How can I access python variable in Spark SQL?
  25. PySpark: How to fillna values in dataframe for specific columns?
  26. How to exclude multiple columns in Spark dataframe in Python
  27. Stratified sampling with pyspark
  28. How to turn off scientific notation in pyspark?
  29. spark.ml StringIndexer throws ‘Unseen label’ on fit()
  30. PySpark error: AttributeError: ‘NoneType’ object has no attribute ‘_jvm’

Leave a Comment