pyspark: count distinct over a window

by

EDIT: as noleto mentions in his answer below, there is now approx_count_distinct available since PySpark 2.1 that works over a window.

Original answer – exact distinct count (not an approximation)

We can use a combination of size and collect_set to mimic the functionality of countDistinct over a window:

from pyspark.sql import functions as F, Window

# Function to calculate number of seconds from number of days
days = lambda i: i * 86400

# Create some test data
df = spark.createDataFrame([(17, "2017-03-10T15:27:18+00:00", "orange"),
                    (13, "2017-03-15T12:27:18+00:00", "red"),
                    (25, "2017-03-18T11:27:18+00:00", "red")],
                    ["dollars", "timestampGMT", "color"])
       
# Convert string timestamp to timestamp type             
df = df.withColumn('timestampGMT', df.timestampGMT.cast('timestamp'))

# Create window by casting timestamp to long (number of seconds)
w = Window.orderBy(F.col("timestampGMT").cast('long')).rangeBetween(-days(7), 0)

# Use collect_set and size functions to perform countDistinct over a window
df = df.withColumn('distinct_color_count_over_the_last_week', F.size(F.collect_set("color").over(w)))

df.show()

This results in the distinct count of color over the previous week of records:

+-------+--------------------+------+---------------------------------------+
|dollars|        timestampGMT| color|distinct_color_count_over_the_last_week|
+-------+--------------------+------+---------------------------------------+
|     17|2017-03-10 15:27:...|orange|                                      1|
|     13|2017-03-15 12:27:...|   red|                                      2|
|     25|2017-03-18 11:27:...|   red|                                      1|
+-------+--------------------+------+---------------------------------------+

More Related Contents:

  1. Avoid performance impact of a single partition mode in Spark window functions
  2. How to melt Spark DataFrame?
  3. Using a column value as a parameter to a spark DataFrame function
  4. Unpivot in spark-sql/pyspark
  5. Split Spark Dataframe string column into multiple columns
  6. Spark Window Functions – rangeBetween dates
  7. Spark Dataframe validating column names for parquet writes
  8. How to check if spark dataframe is empty?
  9. Spark DataFrame: count distinct values of every column
  10. How to split a list to multiple columns in Pyspark?
  11. pyspark dataframe filter or include based on list
  12. How to save/insert each DStream into a permanent table
  13. Pyspark : forward fill with last observation for a DataFrame
  14. How to loop through each row of dataFrame in pyspark
  15. Fill in null with previously known good value with pyspark
  16. Why does Spark think this is a cross / Cartesian join
  17. PySpark – get row number for each row in a group
  18. How to find count of Null and Nan values for each column in a PySpark dataframe efficiently?
  19. Rename more than one column using withColumnRenamed
  20. Keep only duplicates from a DataFrame regarding some field
  21. How to aggregate over rolling time window with groups in Spark
  22. Using windowing functions in Spark
  23. How can I access python variable in Spark SQL?
  24. PySpark: How to fillna values in dataframe for specific columns?
  25. pyspark: rolling average using timeseries data
  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