DataFrame Spark принимает (MAX-MIN) для каждой группы

Question

у меня есть кадр данных из обрабатывающей части, выглядит следующим образом:

   +---------+------+-----------+
|Time     |group |value      |
+---------+------+-----------+
|    28371|    94|        906|
|    28372|    94|        864|
|    28373|    94|        682|
|    28374|    94|        574|
|    28383|    95|        630|
|    28384|    95|        716|
|    28385|    95|        913|

Я хотел бы взять (значение для максимального времени - значение для минимального времени) для каждой группы, чтобы получить этот результат:

+------+-----------+
|group |  value    |
+------+-----------+
|    94|       -332|
|    95|        283|

Заранее благодарю за помощь

Answer 1

df.groupBy("groupCol").agg(max("value")-min("value"))

Исходя из вопроса, отредактированного OP, вот способ сделать это в PySpark.Идея состоит в том, чтобы вычислить номера строк в порядке возрастания и убывания времени для каждой группы и использовать эти значения для вычитания.

from pyspark.sql import Window
from pyspark.sql import functions as func
w_asc = Window.partitionBy(df.groupCol).orderBy(df.time)
w_desc = Window.partitionBy(df.groupCol).orderBy(func.desc(df.time))
df = df.withColumn(func.row_number().over(w_asc).alias('rnum_asc')) \
       .withColumn(func.row_number().over(w_desc).alias('rnum_desc'))
df.groupBy(df.groupCol) \
  .agg((func.max(func.when(df.rnum_desc==1,df.value))-func.max(func.when(df.rnum_asc==1,df.value))).alias('diff')).show()

Было бы проще, если бы в Spark SQL была доступна оконная функция first_value.Общий способ решить эту проблему с помощью SQL -

select distinct groupCol,diff
from (
select t.*
      ,first_value(val) over(partition by groupCol order by time) - 
       first_value(val) over(partition by groupCol order by time desc) as diff
from tbl t
) t