PySpark：使用第一个值“null”复制行

Question

我是 PySpark 的超级新手，我正在尝试获取相同 id 内的值之间的差异。我正在为 DataFrame 使用 csv 格式。

比如我的数据集是这样的：

+---+-----+
| id|value|
+---+-----+
|  1|   65|
|  1|   66|
|  1|   65|
|  2|   68|
|  2|   71|
+---+-----+

我想要这样的东西

+---+-----+----------+
| id|value|prev_value|
+---+-----+----------+
|  1|   65|      null|
|  1|   66|        65|
|  1|   65|        66|
|  2|   68|        65|
|  2|   71|        68|
+---+-----+----------+

这样计算值之间的差异就很容易了。

Answer 1

无法保证顺序，但您可以使用monotonically_increasing_id 和Window 添加行号。如果你能在数据进入spark之前得到数据中的行号，那么就可以保证顺序。获得行号后，您可以加入行号 - 1。

警告：根据上游分区，monotonically_increasing_id 不能保证文件顺序的保留

import pyspark.sql.functions as f
from pyspark.sql.window import Window


# Ideally you could add a partition to avoid performance hit, or have row numbers in the raw file
# e.g. Window.partitionBy(<some partion key>).orderBy(f.monotonically_increasing_id()) 
win_spec = Window.orderBy(f.monotonically_increasing_id())
df = df.withColumn("row_num", f.row_number().over(win_spec))

# Self join on row - 1
df_prev = df.select(col("row_num").alias("prev_row_num"), col("value").alias("prev_value"))
df_res = df.join(df_prev, (df.row_num - 1) == df_prev.prev_row_num, how="left")

df_res.select(["id", "value", "prev_value"]).show()

#  +---+-----+----------+
#  | id|value|prev_value|
#  +---+-----+----------+
#  |  1|   65|      null|
#  |  1|   66|        65|
#  |  1|   67|        66|
#  |  2|   68|        67|
#  |  2|   71|        68|
#  +---+-----+----------+