PySpark: Dataframe Sort Within Partitions

Pradeep

PySpark: Dataframe Sort Within Partitions

This tutorial will explain with examples on how to sort data within partitions based on specified column(s) in a dataframe.

Unlike Sort function, sortWithinPartitions function will not result in shuffle partitions.
By default, sort order within partitions will be ascending if not specified.
Syntax: This function takes 2 parameter, 1st parameter is mandatory but 2nd parameter is optional.
```
sortWithinPartitions(*cols, ascending=True / ascending =[list of 1 and 0])
```
- → 1st parameter is used to specify a column name or list of column names. This parameter can take 3 types of values
  1. String for a column name to sort data.
  2. Comma separated String for column names to sort data.
  3. List of strings of column names in the order in which sorting is required.
- → 2nd parameter can be used to specify order of either single column or list of zeros and ones. This parameter can take 2 types of values
  1. Boolean value, i.e. True or False for single column in 1st parameter
  2. List of booleans to be passed as a key value pair. 1 in list represent ascending and 0 represent descending.
Following topics will be covered on this page, click on item in the below list and it will take you to the respective section of the page:
Sample Data: Dataset used in the below examples can be downloaded from here. Additional column partition# has been added in dataframe to show that sort is happening within partitions.
```
from pyspark.sql.functions import spark_partition_id
df = spark.read.csv("file:///path_to_files/csv_with_duplicates_and_nulls.csv",header=True).repartitionByRange(3,"db_id")
df = df.withColumn("partition#", spark_partition_id())

df.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   12| Teradata|  RDBMS|         0|
|   14|Snowflake|CloudDB|         0|
|   12| Teradata|  RDBMS|         0|
|   15|  Vertica|  RDBMS|         1|
|   22|    Mysql|   null|         1|
|   50|Snowflake|  RDBMS|         2|
|   51|     null|CloudDB|         2|
+-----+---------+-------+----------+
```

➠ Single Column Ascending Order: Using single column to sort data within partitions of a dataframe in an ascending order.

Example 1: Sorting data within partitions of a dataframe based on a single column "db_id" in ascending order(default order).


df_update = df.sortWithinPartitions("db_id")

df_update.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   12| Teradata|  RDBMS|         0|
|   12| Teradata|  RDBMS|         0|
|   14|Snowflake|CloudDB|         0|
|   15|  Vertica|  RDBMS|         1|
|   22|    Mysql|   null|         1|
|   50|Snowflake|  RDBMS|         2|
|   51|     null|CloudDB|         2|
+-----+---------+-------+----------+

Example 2: Sorting data within partitions of a dataframe based on a single column "db_id" in ascending order using asc function.


from pyspark.sql.functions import asc
df_update = df.sortWithinPartitions(asc("db_id"))

df_update.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   12| Teradata|  RDBMS|         0|
|   12| Teradata|  RDBMS|         0|
|   14|Snowflake|CloudDB|         0|
|   15|  Vertica|  RDBMS|         1|
|   22|    Mysql|   null|         1|
|   50|Snowflake|  RDBMS|         2|
|   51|     null|CloudDB|         2|
+-----+---------+-------+----------+

Example 3: Sorting data within partitions of a dataframe based on a single column "db_id" in ascending order using asc column function.


from pyspark.sql.functions import col
df_update = df.sortWithinPartitions(col("db_id").asc())

df_update.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   12| Teradata|  RDBMS|         0|
|   12| Teradata|  RDBMS|         0|
|   14|Snowflake|CloudDB|         0|
|   15|  Vertica|  RDBMS|         1|
|   22|    Mysql|   null|         1|
|   50|Snowflake|  RDBMS|         2|
|   51|     null|CloudDB|         2|
+-----+---------+-------+----------+

➠ Single Column Descending Order: Using single column to sort data within partitions of a dataframe a descending order.

Example 1: Sorting data within partitions of a dataframe based on a single column "db_id" in descending order using ascending parameter.


df_update = df.sortWithinPartitions("db_id", ascending=False)

df_update.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   14|Snowflake|CloudDB|         0|
|   12| Teradata|  RDBMS|         0|
|   12| Teradata|  RDBMS|         0|
|   22|    Mysql|   null|         1|
|   15|  Vertica|  RDBMS|         1|
|   51|     null|CloudDB|         2|
|   50|Snowflake|  RDBMS|         2|
+-----+---------+-------+----------+

Example 2: Sorting data within partitions of a dataframe based on a single column "db_id" in descending order using desc function.


from pyspark.sql.functions import desc
df_update = df.sortWithinPartitions(desc("db_id"))

df_update.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   14|Snowflake|CloudDB|         0|
|   12| Teradata|  RDBMS|         0|
|   12| Teradata|  RDBMS|         0|
|   22|    Mysql|   null|         1|
|   15|  Vertica|  RDBMS|         1|
|   51|     null|CloudDB|         2|
|   50|Snowflake|  RDBMS|         2|
+-----+---------+-------+----------+

Example 3: Sorting data within partitions of a dataframe based on a single column "db_id" in descending order using desc column function.


from pyspark.sql.functions import col
df_update = df.sortWithinPartitions(col("db_id").desc())

df_update.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   14|Snowflake|CloudDB|         0|
|   12| Teradata|  RDBMS|         0|
|   12| Teradata|  RDBMS|         0|
|   22|    Mysql|   null|         1|
|   15|  Vertica|  RDBMS|         1|
|   51|     null|CloudDB|         2|
|   50|Snowflake|  RDBMS|         2|
+-----+---------+-------+----------+

➠ Single Column Ascending Order nulls first: Using single column to sort data within partitions of a dataframe an ascending order with nulls at the beginning.

Example 1: Sorting data within partitions of a dataframe based on a single column "db_name" in ascending order with nulls at the beginning.


from pyspark.sql.functions import col,asc_nulls_first
df_update = df.sortWithinPartitions(asc_nulls_first("db_name"))

df_update.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   14|Snowflake|CloudDB|         0|
|   12| Teradata|  RDBMS|         0|
|   12| Teradata|  RDBMS|         0|
|   22|    Mysql|   null|         1|
|   15|  Vertica|  RDBMS|         1|
|   51|     null|CloudDB|         2|
|   50|Snowflake|  RDBMS|         2|
+-----+---------+-------+----------+



#same result as above
df_update = df.sortWithinPartitions(col("db_name").asc_nulls_first())

➠ Single Column Ascending Order nulls last: Using single column to sort data within partitions of a dataframe an ascending order with nulls at the end.

Example 1: Sorting data within partitions of a dataframe based on a single column "db_name" in ascending order with nulls at the end.


from pyspark.sql.functions import col,asc_nulls_last
df_update = df.sortWithinPartitions(asc_nulls_last("db_name"))

df_update.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   14|Snowflake|CloudDB|         0|
|   12| Teradata|  RDBMS|         0|
|   12| Teradata|  RDBMS|         0|
|   22|    Mysql|   null|         1|
|   15|  Vertica|  RDBMS|         1|
|   50|Snowflake|  RDBMS|         2|
|   51|     null|CloudDB|         2|
+-----+---------+-------+----------+


#same result as above
df_update = df.sortWithinPartitions(col("db_name").asc_nulls_last())

➠ Single Column Descending Order nulls first: Using single column to sort data within partitions of a dataframe a descending order with nulls at the beginning.

Example 1: Sorting data within partitions of a dataframe based on a single column "db_name" in descending order with nulls at the beginning.


from pyspark.sql.functions import col,desc_nulls_first
df_update = df.sortWithinPartitions(desc_nulls_first("db_name"))

df_update.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   12| Teradata|  RDBMS|         0|
|   12| Teradata|  RDBMS|         0|
|   14|Snowflake|CloudDB|         0|
|   15|  Vertica|  RDBMS|         1|
|   22|    Mysql|   null|         1|
|   51|     null|CloudDB|         2|
|   50|Snowflake|  RDBMS|         2|
+-----+---------+-------+----------+


#same result as above
df_update = df.sortWithinPartitions(col("db_name").desc_nulls_first())

➠ Single Column Descending Order nulls last: Using single column to sort data within partitions of a dataframe a descending order with nulls at the end.

Example 1: Sorting data within partitions of a dataframe based on a single column "db_name" in descending order with nulls at the end.


from pyspark.sql.functions import col,desc_nulls_last
df_update = df.sortWithinPartitions(desc_nulls_last("db_name"))

df_update.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   12| Teradata|  RDBMS|         0|
|   12| Teradata|  RDBMS|         0|
|   14|Snowflake|CloudDB|         0|
|   15|  Vertica|  RDBMS|         1|
|   22|    Mysql|   null|         1|
|   50|Snowflake|  RDBMS|         2|
|   51|     null|CloudDB|         2|
+-----+---------+-------+----------+


#same result as above
df_update = df.sortWithinPartitions(col("db_name").desc_nulls_last())

➠ Multiple Columns Ascending Order: Using multi columns "db_name","db_id" to sort data within partitions of a dataframe an ascending order.

Example 1: Sorting data within partitions of a dataframe based on multiple columns "db_name", "db_id" in ascending order(default order).


df_update = df.sortWithinPartitions("db_name","db_id")

df_update.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   14|Snowflake|CloudDB|         0|
|   12| Teradata|  RDBMS|         0|
|   12| Teradata|  RDBMS|         0|
|   22|    Mysql|   null|         1|
|   15|  Vertica|  RDBMS|         1|
|   51|     null|CloudDB|         2|
|   50|Snowflake|  RDBMS|         2|
+-----+---------+-------+----------+

Example 1: Sorting data within partitions of a dataframe based on multiple columns "db_name", "db_id" in ascending order(default order) using list.


sort_list= ["db_name","db_id"]
df_update = df.sortWithinPartitions(sort_list)

df_update.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   14|Snowflake|CloudDB|         0|
|   12| Teradata|  RDBMS|         0|
|   12| Teradata|  RDBMS|         0|
|   22|    Mysql|   null|         1|
|   15|  Vertica|  RDBMS|         1|
|   51|     null|CloudDB|         2|
|   50|Snowflake|  RDBMS|         2|
+-----+---------+-------+----------+

Example 3: Sorting data within partitions of a dataframe based on multiple columns "db_name", "db_id" in ascending order using list with ascending parameter.


df_update = df.sortWithinPartitions("db_name","db_id",ascending=[1,1])

df_update.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   14|Snowflake|CloudDB|         0|
|   12| Teradata|  RDBMS|         0|
|   12| Teradata|  RDBMS|         0|
|   22|    Mysql|   null|         1|
|   15|  Vertica|  RDBMS|         1|
|   51|     null|CloudDB|         2|
|   50|Snowflake|  RDBMS|         2|
+-----+---------+-------+----------+

➠ Multiple Columns Descending Order: Using multiple columns to sort data within partitions of a dataframe a descending order.

Example 1: Sorting data within partitions of a dataframe based on "db_name" and "db_id" columns in descending order.


from pyspark.sql.functions import asc,desc
df_update = df.sortWithinPartitions(desc("db_name"),desc("db_id"))

df_update.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   12| Teradata|  RDBMS|         0|
|   12| Teradata|  RDBMS|         0|
|   14|Snowflake|CloudDB|         0|
|   15|  Vertica|  RDBMS|         1|
|   22|    Mysql|   null|         1|
|   50|Snowflake|  RDBMS|         2|
|   51|     null|CloudDB|         2|
+-----+---------+-------+----------+

Example 2: Sorting data within partitions of a dataframe based on multiple columns "db_name", "db_id" in descending order using list with ascending parameter.


df_update = df.sortWithinPartitions("db_name","db_id",ascending=[0,0])

df_update.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   12| Teradata|  RDBMS|         0|
|   12| Teradata|  RDBMS|         0|
|   14|Snowflake|CloudDB|         0|
|   15|  Vertica|  RDBMS|         1|
|   22|    Mysql|   null|         1|
|   50|Snowflake|  RDBMS|         2|
|   51|     null|CloudDB|         2|
+-----+---------+-------+----------+

➠ Multiple Columns Ascending/Descending(mix) Order: Using multiple columns to sort data within partitions of a dataframe a descending or an ascending order.

Example 1: Sorting data within partitions of a dataframe based on "db_name" in descending order and "db_id" in ascending order.


from pyspark.sql.functions import asc,desc
df_update = df.sortWithinPartitions(desc("db_name"),asc("db_id"))

df_update.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   12| Teradata|  RDBMS|         0|
|   12| Teradata|  RDBMS|         0|
|   14|Snowflake|CloudDB|         0|
|   15|  Vertica|  RDBMS|         1|
|   22|    Mysql|   null|         1|
|   50|Snowflake|  RDBMS|         2|
|   51|     null|CloudDB|         2|
+-----+---------+-------+----------+

Example 2: Sorting data within partitions of a dataframe based on "db_name" in descending order and "db_id" in ascending order using list with ascending parameter.


from pyspark.sql.functions import asc,desc
df_update = df.sortWithinPartitions("db_name","db_id",ascending=[0,1])

df_update.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   12| Teradata|  RDBMS|         0|
|   12| Teradata|  RDBMS|         0|
|   14|Snowflake|CloudDB|         0|
|   15|  Vertica|  RDBMS|         1|
|   22|    Mysql|   null|         1|
|   50|Snowflake|  RDBMS|         2|
|   51|     null|CloudDB|         2|
+-----+---------+-------+----------+

➠ Multiple Columns Ascending/Descending Order nulls first/last: Using single column to sort data within partitions of a dataframe a descending order with nulls at the beginning.

Example 1: Sorting data within partitions of a dataframe based on column "db_type" in ascending order with nulls at the last and column "db_name" in descending order with nulls at the last.


from pyspark.sql.functions import col,desc_nulls_last,asc_nulls_last
df_update = df.sortWithinPartitions(asc_nulls_last("db_type"), desc_nulls_last("db_name"))

df_update.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   14|Snowflake|CloudDB|         0|
|   12| Teradata|  RDBMS|         0|
|   12| Teradata|  RDBMS|         0|
|   15|  Vertica|  RDBMS|         1|
|   22|    Mysql|   null|         1|
|   51|     null|CloudDB|         2|
|   50|Snowflake|  RDBMS|         2|
+-----+---------+-------+----------+


#same result as above
df_update = df.sortWithinPartitions(col("db_type").asc_nulls_last(), col("db_name").desc_nulls_last())

Example 2: Sorting data within partitions of a dataframe based on column "db_type" in ascending order with nulls at the beginning and column "db_name" in descending order with nulls at the last.


from pyspark.sql.functions import col,desc_nulls_last,asc_nulls_first
df_update = df.sortWithinPartitions(asc_nulls_first("db_type"), desc_nulls_last("db_name"))

df_update.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   14|Snowflake|CloudDB|         0|
|   12| Teradata|  RDBMS|         0|
|   12| Teradata|  RDBMS|         0|
|   22|    Mysql|   null|         1|
|   15|  Vertica|  RDBMS|         1|
|   51|     null|CloudDB|         2|
|   50|Snowflake|  RDBMS|         2|
+-----+---------+-------+----------+

Example 3: Sorting data within partitions of a dataframe based on column "db_type" in descending order with nulls at the end and "db_name" in ascending order with nulls at the beginning.


from pyspark.sql.functions import col,desc_nulls_last,asc_nulls_first
df_update = df.sortWithinPartitions(desc_nulls_last("db_type"), asc_nulls_first("db_name"))

df_update.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   12| Teradata|  RDBMS|         0|
|   12| Teradata|  RDBMS|         0|
|   14|Snowflake|CloudDB|         0|
|   15|  Vertica|  RDBMS|         1|
|   22|    Mysql|   null|         1|
|   50|Snowflake|  RDBMS|         2|
|   51|     null|CloudDB|         2|
+-----+---------+-------+----------+

Example 4: Sorting data within partitions of a dataframe based on column "db_type" in ascending order with nulls at the beginning and "db_name" in descending order with nulls at the end.


from pyspark.sql.functions import col,desc_nulls_last,asc_nulls_first
df_update = df.sortWithinPartitions(desc_nulls_first("db_type"), asc_nulls_last("db_name"))

df_update.show()
+-----+---------+-------+----------+
|db_id|  db_name|db_type|partition#|
+-----+---------+-------+----------+
|   12| Teradata|  RDBMS|         0|
|   12| Teradata|  RDBMS|         0|
|   14|Snowflake|CloudDB|         0|
|   22|    Mysql|   null|         1|
|   15|  Vertica|  RDBMS|         1|
|   50|Snowflake|  RDBMS|         2|
|   51|     null|CloudDB|         2|
+-----+---------+-------+----------+

This tutorial will explain with examples on how to sort data within partitions based on specified column(s) in a dataframe.

dbmstutorials.com

PySpark: Dataframe Sort Within Partitions