Блог

У меня есть фрейм данных Spark на Python, и он находится в определенном порядке, в котором строки могут быть разделены на правильные группы в соответствии со столбцом «start_of_section», который имеет значения 1 или 0. Для каждой коллекции строк, которые необходимо сгруппировать вместе, все столбцы, кроме «value» и «start_of_section», равны. Я хочу сгруппировать каждую такую коллекцию в одну строку, которая имеет одинаковые значения для каждого другого столбца, и столбец «list_values», который содержит массив всех значений в каждой строке.

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

 Row(category=fruit, object=apple, value=60, start_of_section=1)
Row(category=fruit, object=apple, value=160, start_of_section=0)
Row(category=fruit, object=apple, value=30, start_of_section=0)
  

и в новом фрейме данных это было бы

 Row(category=fruit, object=apple, list_values=[60, 160, 30])
  

(Редактировать: обратите внимание, что столбец «start_of_section» не должен был быть включен в окончательный фрейм данных.)

Проблема, с которой я столкнулся при попытке найти ответ, заключается в том, что я нашел только способы группировки по значению столбца без учета порядка, так что это привело бы к ошибочному получению двух строк, одна из которых группирует все строки с помощью «start_of_section» = 1, а другая группирует все строки с помощью «start_of_section» = 0..

Какой код может этого достичь?

Предполагая, что ваш столбец order равен order_col

 df.show()
 -------- ------ --------- ---------------- ----- 
|category|object|order_col|start_of_section|value|
 -------- ------ --------- ---------------- ----- 
|   fruit| apple|        1|               1|   60|
|   fruit| apple|        2|               0|  160|
|   fruit| apple|        3|               0|   30|
|   fruit| apple|        4|               1|   50|
 -------- ------ --------- ---------------- ----- 
  

вам нужно сгенерировать идентификатор, чтобы сгруппировать строки в одном разделе вместе, затем сгруппировать по этому идентификатору и нужному размеру. Вот как вы это делаете.

 from pyspark.sql import functions as F, Window as W

df.withColumn(
    "id",
    F.sum("start_of_section").over(
        W.partitionBy("category", "object").orderBy("order_col")
    ),
).groupBy("category", "object", "id").agg(F.collect_list("value").alias("values")).drop(
    "id"
).show()

 -------- ------ ------------- 
|category|object|       values|
 -------- ------ ------------- 
|   fruit| apple|[60, 160, 30]|
|   fruit| apple|         [50]|
 -------- ------ ------------- 
  

РЕДАКТИРОВАТЬ: Если у вас их нет order_col , это невыполнимая задача. Смотрите на свои строки во фрейме данных как на мрамор в пакете. У них нет никакого порядка. Вы можете упорядочивать их по мере извлечения из пакета в соответствии с некоторыми критериями, но в противном случае вы не можете предполагать какой-либо порядок. show вы просто вытаскиваете 10 шариков (строк) из сумки. Порядок может быть одинаковым каждый раз, когда вы это делаете, но внезапно меняться, и вы не можете его контролировать

Ну, теперь я понял. Вы можете выполнить группировку с помощью столбца, суммирующего start_of_section .

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

 from pyspark.sql.types import Row
from pyspark.sql.functions import *
from pyspark.sql import Window

data = [Row(category='fruit', object='apple', value=60, start_of_section=1),
    Row(category='fruit', object='apple', value=160, start_of_section=0),
    Row(category='fruit', object='apple', value=30, start_of_section=0),
    Row(category='fruit', object='apple', value=50, start_of_section=1),
    Row(category='fruit', object='apple', value=30, start_of_section=0),
    Row(category='fruit', object='apple', value=60, start_of_section=1),
    Row(category='fruit', object='apple', value=110, start_of_section=0)]

df = spark.createDataFrame(data)

w = Window.partitionBy('category', 'object').rowsBetween(Window.unboundedPreceding, Window.currentRow)

df.withColumn('group', sum('start_of_section').over(w)) 
  .groupBy('category', 'object', 'group').agg(collect_list('value').alias('list_value')) 
  .drop('group').show()

 -------- ------ ------------- 
|category|object|   list_value|
 -------- ------ ------------- 
|   fruit| apple|[60, 160, 30]|
|   fruit| apple|     [50, 30]|
|   fruit| apple|    [60, 110]|
 -------- ------ ------------- 
  

СБОЙ: monotonically_increasing_id сбой при наличии большого количества разделов.

 df.repartition(7) 
  .withColumn('id', monotonically_increasing_id()) 
  .withColumn('group', sum('start_of_section').over(w)) 
  .groupBy('category', 'object', 'group').agg(collect_list('value').alias('list_value')) 
  .drop('group').show()

 -------- ------ -------------------- 
|category|object|          list_value|
 -------- ------ -------------------- 
|   fruit| apple|                [60]|
|   fruit| apple|[60, 160, 30, 30,...|
|   fruit| apple|                [50]|
 -------- ------ -------------------- 
  

Это совершенно нежелательно.