Сдвиг слотов в фрейме данных с помощью pyspark

Question

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

Cust_id|slot|trigger_id|coup_type
1|       1| 2101| null
1|       2| 2102| null
1|       3| 2103| null
1|       4| 2104| null
1|       5| 2105| product
1|       6| 2106| null
1|       7| 2107| null
1|       8| 2108| product
1|       9| 2109| null
1|       10| 21010| null
1|       11| 21011| product
1|       12| 21012| product

Теперь мне нужно выполнить смещение слота на основе coup_type, чтобы все перевороты продукта присутствовали в слоте 7-10, а выделение всегда должно начинаться с слота 7. после переключения слот следует переставить, как показано ниже:-

Cust_id|slot|trigger_id|coup_type
1|       1| 2101| null
1|       2| 2102| null
1|       3| 2103| null
1|       4| 2104| null
1|       5| 2105| null
1|       6| 2106| null
1|       7| 2105| product
1|       8| 2108| product
1|       9| 21011| product
1|       10| 21012| product
1|       11| 21009| null
1|       12| 21010| null

Мне нужно выполнить эту операцию в pyspark.

Пожалуйста, дайте мне знать, если вопрос не ясен.

Заранее спасибо.

Answer 1

Вы можете получить row_number заказав колонку coup_type и сделайте его номером строки как ключ соединения с slotстолбец. Если столбец не является непрерывным, вам может потребоваться другой столбец для номера строки заказаslot столбец.

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

w = Window.partitionBy('Cust_id').orderBy(desc('coup_type'))

df2 = df.withColumn('slot', row_number().over(w) + 7 - 1).drop('trigger_id')
df2.show(12, False)

+-------+----+---------+
|Cust_id|slot|coup_type|
+-------+----+---------+
|1      |7   |product  |
|1      |8   |product  |
|1      |9   |product  |
|1      |10  |product  |
|1      |11  |null     |
|1      |12  |null     |
|1      |13  |null     |
|1      |14  |null     |
|1      |15  |null     |
|1      |16  |null     |
|1      |17  |null     |
|1      |18  |null     |
+-------+----+---------+

df.drop('coup_type').join(df2, ['Cust_id', 'slot'], 'left').show(12, False)

+-------+----+----------+---------+
|Cust_id|slot|trigger_id|coup_type|
+-------+----+----------+---------+
|1      |1   |2101      |null     |
|1      |2   |2102      |null     |
|1      |3   |2103      |null     |
|1      |4   |2104      |null     |
|1      |5   |2105      |null     |
|1      |6   |2106      |null     |
|1      |7   |2107      |product  |
|1      |8   |2108      |product  |
|1      |9   |2109      |product  |
|1      |10  |21010     |product  |
|1      |11  |21011     |null     |
|1      |12  |21012     |null     |
+-------+----+----------+---------+