python - PySpark:如何根据其他行值的值更改行+列的值

Question

我有一个像这样的数据框...

+----------+-----+
|      date|price|
+----------+-----+
|2019-01-01|   25|
|2019-01-02|   22|
|2019-01-03|   20|
|2019-01-04|   -5|
|2019-01-05|   -1|
|2019-01-06|   -2|
|2019-01-07|    5|
|2019-01-08|  -11|
+----------+-----+

我想基于需要回顾其他行的逻辑创建一个新列 - 而不仅仅是同一行的列值

我正在尝试一些 UDF，但它采用列的相应行值。我不知道如何查看其他行...

举例:我想创建一个新列“newprice” - 类似于这样......

+----------+-----+----------+
|      date|price|new price
+----------+-----+----------+
|2019-01-01|   25| 25
|2019-01-02|   22| 22
|2019-01-03|   20| 20
|2019-01-04|   -5| 20
|2019-01-05|   -1| 20
|2019-01-06|   -2| 20
|2019-01-07|    5| 5
|2019-01-08|  -11| 5
+----------+-----+-----------+

本质上，新列值中的每一行都不是基于相应行的值，而是基于其他行的值...

逻辑:如果价格为负，则回顾前几天，如果当天为正值 - 接受它或再返回一天，直到出现正值...

    dateprice = [('2019-01-01',25),('2019-01-02',22),('2019-01-03',20),('2019-01-04', -5),\
     ('2019-01-05',-1),('2019-01-06',-2),('2019-01-07',5),('2019-01-08', -11)]

   dataDF = sqlContext.createDataFrame(dateprice, ('date', 'price'))

任何帮助将不胜感激。

Answer 1

首先使用 price 列填充new Price 列，但将负值替换为 null。然后您可以使用 Fill in null with previously known good value with pyspark 上显示的技术获取最后一个非空值，在本例中这将是最后一个正值。

例如:

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

w = Window.orderBy("date").rowsBetween(Window.unboundedPreceding, Window.currentRow)

dataDF.withColumn("new_price", when(col("price") >= 0, col("price")))\
    .withColumn(
        "new_price",
        last('new_price', True).over(w)
    )\
    .show()
#+----------+-----+---------+
#|      date|price|new_price|
#+----------+-----+---------+
#|2019-01-01|   25|       25|
#|2019-01-02|   22|       22|
#|2019-01-03|   20|       20|
#|2019-01-04|   -5|       20|
#|2019-01-05|   -1|       20|
#|2019-01-06|   -2|       20|
#|2019-01-07|    5|        5|
#|2019-01-08|  -11|        5|
#+----------+-----+---------+

在这里我利用了when这一事实如果条件不匹配且未指定 otherwise，则默认返回 null。