Группировка значений в Pyspark

Question

Мне нужна помощь со следующим.Предположим, у меня есть фрейм данных, как показано ниже.Я пытаюсь сгенерировать приложение столбцов dom и attribute на основе порядка столбца «rank».

from pyspark.sql.functions import first
l =[( 1    ,'A', 10, 'size' ),
( 2        , 'B', 20, 'height' ),
( 3        , 'C', 30, 'weigth' )]

df = spark.createDataFrame(l, ['rank','dom', 'value', 'attribute'])

+----+---+-----+---------+
|rank|dom|value|attribute|
+----+---+-----+---------+
|   1|  A|   10|     size|
|   2|  B|   20|   height|
|   3|  C|   30|   weigth|
+---+---+-----+---------+

Окончательный желаемый результат:

+----+-------+---------+--------------------+
|rank|    dom|avg_value|           attribute|
+----+-------+---------+--------------------+
|   1|      A|       10|                size|
|   2|    A,B|       20|        size, height|
|   3|  A,B,C|       30|size, height, weigth|
+----+-------+---------+--------------------+

Идея состоит в том,это avg_price - среднее значение, сгруппированное по новой комбинации атрибутов.

Answer 1

вы можете использовать функциональность окна и собирать предыдущие значения атрибутов.

import pyspark.sql.functions as f
from pyspark.sql.window import Window

l =[( 1    ,'A', 10, 'size' ),
( 2        , 'B', 20, 'height' ),
( 3        , 'C', 30, 'weigth' )]

df = spark.createDataFrame(l, ['rank','dom', 'value', 'attribute'])

windowSpec = Window().orderBy('rank').rowsBetween(Window.unboundedPreceding, Window.currentRow)
df = df.withColumn('attribute',f.collect_list(f.col('attribute')).over(windowSpec))
df.show()

Вывод>

+----+---+-----+--------------------+
|rank|dom|value|           attribute|
+----+---+-----+--------------------+
|   1|  A|   10|              [size]|
|   2|  B|   20|      [size, height]|
|   3|  C|   30|[size, height, we...|
+----+---+-----+--------------------+

Если у вас разные группы данных, вы можете создать секционированное окно

windowSpec = Window().partitionBy(PARTITION_COLUMN).orderBy('rank').rowsBetween(Window.unboundedPreceding, Window.currentRow)