Spark SQL Row_number() PartitionBy Сортировка Desc

Я успешно создал row_number() partitionBy в Spark, используя Window, но хотел бы отсортировать его по убыванию, а не по умолчанию. Вот мой рабочий код:

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()

Это дает мне такой результат:

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

И здесь я добавляю desc() для упорядочения по убыванию:

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

И получите эту ошибку:

AttributeError: объект "WindowSpec" не имеет атрибута "desc"

Что я здесь делаю неправильно?

Ответы

Ответ 1

desc должен применяться к столбцу, а не к определению окна. Вы можете использовать либо метод в столбце:

from pyspark.sql.functions import col  

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

или автономной функции:

from pyspark.sql.functions import desc

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

Ответ 2

Или вы можете использовать код SQL в 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()

Ответ 3

Обновление На самом деле, я попытался разобраться в этом, и это, похоже, не работает. (на самом деле это выдает ошибку). Причина, по которой это не сработало, заключается в том, что у меня был этот код при вызове display() в Databricks (код после вызова display() никогда не запускался). Похоже, что orderBy() в кадре данных и orderBy() в window на самом деле не совпадают. Я буду держать этот ответ только для отрицательного подтверждения

Начиная с PySpark 2.4 (и, возможно, ранее), простое добавление ключевого слова ascending=False в вызов orderBy работает для меня.

Ex.

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

и

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

кажется, дают мне такое же поведение.