python - 我们可以动态检索 pyspark 数据框中更新列的前一行值吗

Question

这是我的确切问题陈述。我在下面提到了数据框。

+--------+-------+
|  START |  END  |
+--------+-------+
|   1    |   5   |
|   3    |   6   |
|   7    |   10  |
|   13   |   17  |
|   15   |   20  |
+--------+-------+

假设每行代表 X 轴上从 START 开始到 END 结束的一条线。当我们根据给定的数据放置它们时，我们不希望线条相交。所以我们把它们堆叠起来。

所以第一行保持不变，即 (1, 5)

由于第二行与第一行相交，我们需要更改其 START 和 END 值。因此从 (3, 6) 变为 (5, 8)。 (堆叠时我们无法改变线的长度)

第三行 (7, 10) 变为 (8, 11)(因为它与前一行 (5, 8) 相交)。

由于第四行与更新后的第三行不相交，因此我们不更改其值。所以它仍然是 (13, 17)

最后一个 (15, 20) 变为 (17, 22)。

所以我的最终数据框应该是:

+--------+-------+
|  START |  END  |
+--------+-------+
|   1    |   5   |
|   5    |   8   |
|   8    |   11  |
|   13   |   17  |
|   17   |   22  |
+--------+-------+

您可以认为初始数据帧是按其 START 列排序的。

现在，当我们使用循环时，这是一个简单的问题，但我想在 pyspark 中做到这一点，而不使用任何循环。我是 pyspark 的新手，找不到一个好的函数，这样我就可以在没有循环的情况下实现它。

现在回到我的标题，如果我能够动态检索上一行的 END 值(当它发生变化时)并与当前的 START 值进行比较，我可以解决这个问题。但我找不到任何可以做到这一点的东西。

这是我在没有循环的情况下解决这个问题的尝试:

from pyspark import SparkContext 
from pyspark.sql import SQLContext
from pyspark.sql.functions import lag, col
from pyspark.sql.window import Window
from pyspark.sql import functions as F

sc = SparkContext('local', 'Example_2')
sqlcontext = SQLContext(sc)

df = sqlcontext.createDataFrame([(1, 5), (3, 6), (7, 10), (13, 17), (15, 20)], ['START', 'END'])

w = Window().orderBy('START').rangeBetween(Window.unboundedPreceding, -1)

# Updating 'END' column first
df = df.withColumn('END', F.when(
                F.last('END', True).over(w) > col('START'),
                col('END') + (F.last('END').over(w) - col('START'))
            ).otherwise(col('END')))

# Updating 'START' column
df = df.withColumn('START', F.when(
                    F.last('END', True).over(w) > col('START'),
                    F.last('END', True).over(w)
                ).otherwise(col('START')))

由于 F.last('END') 没有给出更新的最终值，因此上面的代码返回以下内容，其中第三行是错误的。

+-----+---+
|START|END|
+-----+---+
|    1|  5|
|    5|  8|
|    8| 10|
|   13| 17|
|   17| 22|
+-----+---+

Answer 1

您需要使用rowsBetween代替rangeBetween。所以这实际上会选择窗口规范中的行。看到这里https://spark.apache.org/docs/latest/api/python/pyspark.sql.html?highlight=exception#pyspark.sql.Window.rowsBetween

w = Window().orderBy('START').rowsBetween(Window.unboundedPreceding, -1)