Добавление колонки с последним обновлением в PySpark

Question

У меня есть фрейм данных pyspark со списком клиентов, дней и типов транзакций.

+----------+-----+------+
| Customer | Day | Type |
+----------+-----+------+
| A        |   2 | X11  |
| A        |   4 | X2   |
| A        |   9 | Y4   |
| A        |  11 | X1   |
| B        |   3 | Y4   |
| B        |   7 | X1   |
+----------+-----+------+

Я хотел бы создать столбец, который имеет «самый последний тип X» для каждого клиента, например:

+----------+-----+------+-------------+
| Customer | Day | Type | MostRecentX |
+----------+-----+------+-------------+
| A        |   2 | X11  | X11         |
| A        |   4 | X2   | X2          |
| A        |   9 | Y4   | X2          |
| A        |  11 | X1   | X1          |
| B        |   3 | Y4   | -           |
| B        |   7 | X1   | X1          |
+----------+-----+------+-------------+

Таким образом, для типов X он просто берет один из текущей строки, но для типа Y он берет тип из самой последней строки X для этого члена (и если его нет, он получает пустой или что-то). Я предполагаю, что мне нужна своего рода оконная функция, но я не очень знаком с PySpark.

Answer 1

Вы можете получить, взяв в этом столбце last, что startswith букву "X" над Window, которая разбивает на Customer и заказывает поDay.Укажите Window для начала в начале раздела и остановки в текущей строке.

from pyspark.sql import Window
from pyspark.sql.functions import col, last, when

w = Window.partitionBy("Customer").orderBy("Day").rowsBetween(Window.unboundedPreceding, 0)

df = df.withColumn(
    "MostRecentX",
    last(when(col("Type").startswith("X"), col("Type")), ignorenulls=True).over(w)
)
df.show()
#+--------+---+----+-----------+
#|Customer|Day|Type|MostRecentX|
#+--------+---+----+-----------+
#|       A|  2| X11|        X11|
#|       A|  4|  X2|         X2|
#|       A|  9|  Y4|         X2|
#|       A| 11|  X1|         X1|
#|       B|  3|  Y4|       null|
#|       B|  7|  X1|         X1|
#+--------+---+----+-----------+

Хитрость здесь в том, чтобы использовать when для возврата Typeстолбец, только если он начинается с "X".По умолчанию when вернет null.Затем мы можем использовать last с ignorenulls=True, чтобы получить значение для MostRecentX.

Если вы хотите заменить null на "-", как показано в вашем вопросе, просто позвоните fillnaв столбце MostRecentX:

df.fillna("-", subset=["MostRecentX"]).show()
#+--------+---+----+-----------+
#|Customer|Day|Type|MostRecentX|
#+--------+---+----+-----------+
#|       A|  2| X11|        X11|
#|       A|  4|  X2|         X2|
#|       A|  9|  Y4|         X2|
#|       A| 11|  X1|         X1|
#|       B|  3|  Y4|          -|
#|       B|  7|  X1|         X1|
#+--------+---+----+-----------+