collect_list by preserving order based on another variable

by 
from pyspark.sql import functions as F
from pyspark.sql import Window

w = Window.partitionBy('id').orderBy('date')

sorted_list_df = input_df.withColumn(
            'sorted_list', F.collect_list('value').over(w)
        )\
        .groupBy('id')\
        .agg(F.max('sorted_list').alias('sorted_list'))

Window examples provided by users often don’t really explain what is going on so let me dissect it for you.

As you know, using collect_list together with groupBy will result in an unordered list of values. This is because depending on how your data is partitioned, Spark will append values to your list as soon as it finds a row in the group. The order then depends on how Spark plans your aggregation over the executors.

A Window function allows you to control that situation, grouping rows by a certain value so you can perform an operation over each of the resultant groups:

w = Window.partitionBy('id').orderBy('date')
  • partitionBy – you want groups/partitions of rows with the same id
  • orderBy – you want each row in the group to be sorted by date

Once you have defined the scope of your Window – “rows with the same id, sorted by date” -, you can use it to perform an operation over it, in this case, a collect_list:

F.collect_list('value').over(w)

At this point you created a new column sorted_list with an ordered list of values, sorted by date, but you still have duplicated rows per id. To trim out the duplicated rows you want to groupBy id and keep the max value in for each group:

.groupBy('id')\
.agg(F.max('sorted_list').alias('sorted_list'))

More Related Contents:

  1. Spark Equivalent of IF Then ELSE
  2. How do I add a new column to a Spark DataFrame (using PySpark)?
  3. How to perform union on two DataFrames with different amounts of columns in spark?
  4. Does spark predicate pushdown work with JDBC?
  5. Explode in PySpark
  6. Python worker failed to connect back
  7. How can we JOIN two Spark SQL dataframes using a SQL-esque “LIKE” criterion?
  8. Spark RDD to DataFrame python
  9. Filtering a Pyspark DataFrame with SQL-like IN clause
  10. Spark union of multiple RDDs
  11. Dividing complex rows of dataframe to simple rows in Pyspark
  12. Spark DataFrame TimestampType – how to get Year, Month, Day values from field?
  13. Spark add new column to dataframe with value from previous row
  14. Spark groupByKey alternative
  15. Tuning parameters for implicit pyspark.ml ALS matrix factorization model through pyspark.ml CrossValidator
  16. How to create a custom Estimator in PySpark
  17. Pyspark: Replacing value in a column by searching a dictionary
  18. Spark iteration time increasing exponentially when using join
  19. Pivot String column on Pyspark Dataframe
  20. Reduce a key-value pair into a key-list pair with Apache Spark
  21. Getting Spark, Python, and MongoDB to work together
  22. How do I set the driver’s python version in spark?
  23. Total size of serialized results of 16 tasks (1048.5 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)
  24. PySpark in iPython notebook raises Py4JJavaError when using count() and first()
  25. pyspark parse fixed width text file
  26. Concatenating string by rows in pyspark
  27. PySpark: match the values of a DataFrame column against another DataFrame column
  28. How to process RDDs using a Python class?
  29. What is the Spark DataFrame method `toPandas` actually doing?
  30. how to add Row id in pySpark dataframes [duplicate]

Leave a Comment