Pyspark：使用 DataFrame 的最后一次观察进行前向填充

使用 Spark 1.5.1，

我一直在尝试转发填充空值与最后已知的观察一栏我的数据框。

可以从空值开始，对于这种情况，我将用第一个已知的观察值向后填充该空值。但是，如果这使代码过于复杂，则可以跳过这一点。

In this post，Scala 中的解决方案针对非常相似的问题提供了zero323.

但是，我不了解 Scala，而且我没有成功地在 Pyspark API 代码中“翻译”它。用 Pyspark 可以做到吗？

感谢您的帮助。

下面是一个简单的示例输入：

| cookie_ID     | Time       | User_ID   
| ------------- | --------   |------------- 
| 1             | 2015-12-01 | null 
| 1             | 2015-12-02 | U1
| 1             | 2015-12-03 | U1
| 1             | 2015-12-04 | null   
| 1             | 2015-12-05 | null     
| 1             | 2015-12-06 | U2
| 1             | 2015-12-07 | null
| 1             | 2015-12-08 | U1
| 1             | 2015-12-09 | null      
| 2             | 2015-12-03 | null     
| 2             | 2015-12-04 | U3
| 2             | 2015-12-05 | null   
| 2             | 2015-12-06 | U4

以及预期的输出：

| cookie_ID     | Time       | User_ID   
| ------------- | --------   |------------- 
| 1             | 2015-12-01 | U1
| 1             | 2015-12-02 | U1
| 1             | 2015-12-03 | U1
| 1             | 2015-12-04 | U1
| 1             | 2015-12-05 | U1
| 1             | 2015-12-06 | U2
| 1             | 2015-12-07 | U2
| 1             | 2015-12-08 | U1
| 1             | 2015-12-09 | U1
| 2             | 2015-12-03 | U3
| 2             | 2015-12-04 | U3
| 2             | 2015-12-05 | U3
| 2             | 2015-12-06 | U4

另一种解决方法是尝试这样的方法：

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

window = (
    Window
    .partitionBy('cookie_id')
    .orderBy('Time')
    .rowsBetween(Window.unboundedPreceding, Window.currentRow)
)

final = (
    joined
    .withColumn('UserIDFilled', F.last('User_ID', ignorenulls=True).over(window))
)

因此，它所做的就是根据分区键和顺序列构建窗口。它还告诉窗口回顾窗口内的所有行直到当前行。最后，在每一行，您返回最后一个不为空的值（记住，根据您的窗口，它包括您当前的行）