Pyspark - 涉及最后一天的“递归”函数

Question

我正在 pyspark 中处理一个流程，我有一个数据框，我正在尝试再添加一列（使用 withColumn 方法）。

问题在于公式是：

STATUS1 = 如果 'PETP-today' > 0 则 'Status1 last day' + 'PETP-today' else 0

Status1 的每个结果都包含上一天结果的 status1。

我找到的一个解决方案是创建一个 pandas 数据框并逐个运行记录，直到我可以使用变量计算每个记录。但是我会有性能问题。你能帮忙吗？

考虑数据框列：日期（每日）/PETP（浮动）/STATUS1？ （浮动）

非常感谢任何帮助！

Answer 1

我认为解决方案的关键是 lag 函数。试试这个（为简单起见，我假设所有列的数据都是整数）：

首先，将列向上移动一天

import pyspark
from pyspark.sql import SparkSession
from pyspark import SparkContext
import pandas as pd
from pyspark.sql import functions as F
from pyspark.sql import Window

sc = SparkContext.getOrCreate()
spark = SparkSession(sc)

columns = ['date', 'petp', 'status']
data = [(0, 0, 0), (1, 1, 1), (2, 2, 2), (3,3,3), (4,4,4), (5,5,5)]
pd_data = pd.DataFrame.from_records(data=data, columns=columns)
spark_data = spark.createDataFrame(pd_data)

spark_data_with_lag = spark_data.withColumn("status_last_day", F.lag("status", 1, 0).over(Window.orderBy("date")))
spark_data_with_lag.show()

+----+----+------+---------------+
|date|petp|status|status_last_day|
+----+----+------+---------------+
|   1|   1|     1|              0|
|   2|   2|     2|              1|
|   3|   3|     3|              2|
|   4|   4|     4|              3|
|   5|   5|     5|              4|
+----+----+------+---------------+

然后在条件中使用该数据

status2 = spark_data_with_lag.withColumn("status2", F.when(F.col("date") > 0, F.col("petp") + F.col("status_last_day")).otherwise(0))
status2.show()

+----+----+------+---------------+-------+
|date|petp|status|status_last_day|status2|
+----+----+------+---------------+-------+
|   1|   1|     1|              0|      1|
|   2|   2|     2|              1|      3|
|   3|   3|     3|              2|      5|
|   4|   4|     4|              3|      7|
|   5|   5|     5|              4|      9|
+----+----+------+---------------+-------+

我希望这就是你想要的。