Spark SQL Row_number() PartitionBy Sort Desc

Question

I've successfully create a row_number() partitionBy by in Spark using Window, but would like to sort this by descending, instead of the default ascending. Here is my working code:

from pyspark import HiveContext
from pyspark.sql.types import *
from pyspark.sql import Row, functions as F
from pyspark.sql.window import Window

data_cooccur.select("driver", "also_item", "unit_count", 
    F.rowNumber().over(Window.partitionBy("driver").orderBy("unit_count")).alias("rowNum")).show()

That gives me this result:

 +------+---------+----------+------+
 |driver|also_item|unit_count|rowNum|
 +------+---------+----------+------+
 |   s10|      s11|         1|     1|
 |   s10|      s13|         1|     2|
 |   s10|      s17|         1|     3|

And here I add the desc() to order descending:

data_cooccur.select("driver", "also_item", "unit_count", F.rowNumber().over(Window.partitionBy("driver").orderBy("unit_count").desc()).alias("rowNum")).show()

And get this error:

AttributeError: 'WindowSpec' object has no attribute 'desc'

What am I doing wrong here?

On my PySpark (2.2.0) I have to use `row_number` instead of `rowNumber`. — lambruscoAcido, Jan 25 '18 at 11:34

score 120 · Accepted Answer · edited Jul 13 '20 at 11:51

120

desc should be applied on a column not a window definition. You can use either a method on a column:

from pyspark.sql.functions import col, row_number
from pyspark.sql.window import Window

F.row_number().over(
    Window.partitionBy("driver").orderBy(col("unit_count").desc())
)

or a standalone function:

from pyspark.sql.functions import desc
from pyspark.sql.window import Window

F.row_number().over(
    Window.partitionBy("driver").orderBy(desc("unit_count"))
)

edited Jul 13 '20 at 11:51

jka.ne

1,584
1
15
19

answered Feb 07 '16 at 03:35

zero323

322,348
103
959
935

2

strange the pyspark `orderBy` is different from the `window.orderBy` as one accepts an ascending and the other doesn't. – Umar.H Nov 17 '20 at 16:52

score 4 · Answer 2 · answered May 15 '19 at 18:31

Or you can use the SQL code in Spark-SQL:

from pyspark.sql import SparkSession

spark = SparkSession\
    .builder\
    .master('local[*]')\
    .appName('Test')\
    .getOrCreate()

spark.sql("""
    select driver
        ,also_item
        ,unit_count
        ,ROW_NUMBER() OVER (PARTITION BY driver ORDER BY unit_count DESC) AS rowNum
    from data_cooccur
""").show()

score 1 · Answer 3 · edited Nov 14 '22 at 17:16

1

import pyspark
from pyspark.sql.functions import desc
df.orderBy(desc("col"))

edited Nov 14 '22 at 17:16

jmoerdyk

5,544
7
38
49

answered Nov 14 '22 at 12:03

Karthik Swarna

11
1

information_interchange · Answer 4 · 2019-08-26T19:30:07.773

Update Actually, I tried looking more into this, and it appears to not work. (in fact it throws an error). The reason why it didn't work is that I had this code under a call to display() in Databricks (code after the display() call is never run). It seems like the orderBy() on a dataframe and the orderBy() on a window are not actually the same. I will keep this answer up just for negative confirmation

~~As of PySpark 2.4,(and probably earlier), simply adding in the keyword ascending=False into the orderBy call works for me.~~

Ex.

personal_recos.withColumn("row_number", F.row_number().over(Window.partitionBy("COLLECTOR_NUMBER").orderBy("count", ascending=False)))

and

personal_recos.withColumn("row_number", F.row_number().over(Window.partitionBy("COLLECTOR_NUMBER").orderBy(F.col("count").desc())))

seem to give me the same behaviour.

score 0 · Answer 5 · answered Jun 23 '22 at 20:35

from pyspark.sql.functions import col, row_number
from pyspark.sql.window import Window

my_new_df = df.select(df["STREET NAME"]).distinct()

# Count the rows in my_new_df 
print("\nThere are %d rows in the my_new_df DataFrame.\n" % my_new_df .count())

# Add a ROW_ID
my_new_df = my_new_df .withColumn('ROW_ID', F.monotonically_increasing_id())

# Show the rows with 10 highest IDs in the set and sort DESC
my_new_df .orderBy(my_new_df .ROW_ID.desc()).show(10)

score 0 · Answer 6 · answered Oct 13 '22 at 17:56

Another option, similar to @zero333's col option is using sorting on the column.

data_cooccur.select("driver", "also_item", "unit_count", F.rowNumber().over(Window.partitionBy("driver").orderBy(data_cooccur["unit_count"].desc())).alias("rowNum")).show()

Spark SQL Row_number() PartitionBy Sort Desc

6 Answers6

Linked