Подтвердить что ты не робот

Spark добавляет новый столбец в dataframe со значением из предыдущей строки

Мне интересно, как я могу добиться следующего в Spark (Pyspark)

Начальный Dataframe:

+--+---+
|id|num|
+--+---+
|4 |9.0|
+--+---+
|3 |7.0|
+--+---+
|2 |3.0|
+--+---+
|1 |5.0|
+--+---+

Результирующий Dataframe:

+--+---+-------+
|id|num|new_Col|
+--+---+-------+
|4 |9.0|  7.0  |
+--+---+-------+
|3 |7.0|  3.0  |
+--+---+-------+
|2 |3.0|  5.0  |
+--+---+-------+

Мне удается в целом "добавить" новые столбцы в фреймворк данных, используя что-то вроде: df.withColumn("new_Col", df.num * 10)

Однако я не знаю, как я могу добиться этого "сдвига строк" ​​для нового столбца, так что новый столбец имеет значение поля из предыдущей строки (как показано в примере). Я также не мог найти ничего в документации API о том, как получить доступ к определенной строке в DF по индексу.

Любая помощь будет оценена.

4b9b3361

Ответ 1

Вы можете использовать функцию окна lag следующим образом

from pyspark.sql.functions import lag, col
from pyspark.sql.window import Window

df = sc.parallelize([(4, 9.0), (3, 7.0), (2, 3.0), (1, 5.0)]).toDF(["id", "num"])
w = Window().partitionBy().orderBy(col("id"))
df.select("*", lag("num").over(w).alias("new_col")).na.drop().show()

## +---+---+-------+
## | id|num|new_col|
## +---+---+-------|
## |  2|3.0|    5.0|
## |  3|7.0|    3.0|
## |  4|9.0|    7.0|
## +---+---+-------+

но есть некоторые важные проблемы:

  • Если вам нужна глобальная операция (не секционированная каким-либо другим столбцом/столбцами), она крайне неэффективна.
  • Вам нужен естественный способ заказать ваши данные.

В то время как вторая проблема почти никогда не является проблемой, первая из них может быть разрывом. Если это так, вы должны просто преобразовать ваш DataFrame в RDD и вычислить lag вручную. См. Например:

Другие полезные ссылки: