它使用last并忽略空值。

让我们重新创建与原始数据类似的内容：

import sys
from pyspark.sql.window import Window
import pyspark.sql.functions as func

d = [{'session': 1, 'ts': 1}, {'session': 1, 'ts': 2, 'id': 109}, {'session': 1, 'ts': 3}, {'session': 1, 'ts': 4, 'id': 110}, {'session': 1, 'ts': 5},  {'session': 1, 'ts': 6}]
df = spark.createDataFrame(d)

df.show()

# +-------+---+----+

# |session| ts|  id|

# +-------+---+----+

# |      1|  1|null|

# |      1|  2| 109|

# |      1|  3|null|

# |      1|  4| 110|

# |      1|  5|null|

# |      1|  6|null|

# +-------+---+----+

现在，让我们使用窗口函数last：

df.withColumn("id", func.last('id', True).over(Window.partitionBy('session').orderBy('ts').rowsBetween(-sys.maxsize, 0))).show()

# +-------+---+----+

# |session| ts|  id|

# +-------+---+----+

# |      1|  1|null|

# |      1|  2| 109|

# |      1|  3| 109|

# |      1|  4| 110|

# |      1|  5| 110|

# |      1|  6| 110|

# +-------+---+----+

这似乎是在使用Window functions：

import sys
from pyspark.sql.window import Window
import pyspark.sql.functions as func

def fill_nulls(df):
    df_na = df.na.fill(-1)
    lag = df_na.withColumn('id_lag', func.lag('id', default=-1)
                           .over(Window.partitionBy('session')
                                 .orderBy('timestamp')))

    switch = lag.withColumn('id_change',
                            ((lag['id'] != lag['id_lag']) &
                             (lag['id'] != -1)).cast('integer'))

    switch_sess = switch.withColumn(
        'sub_session',
        func.sum("id_change")
        .over(
            Window.partitionBy("session")
            .orderBy("timestamp")
            .rowsBetween(-sys.maxsize, 0))
    )

    fid = switch_sess.withColumn('nn_id',
                           func.first('id')
                           .over(Window.partitionBy('session', 'sub_session')
                                 .orderBy('timestamp')))

    fid_na = fid.replace(-1, 'null')

    ff = fid_na.drop('id').drop('id_lag')
                          .drop('id_change')
                          .drop('sub_session').
                          withColumnRenamed('nn_id', 'id')

    return ff

下面是完整的null_test.py。

@Oleksiy的答案很好，但并不完全符合我的要求。在一个会话中，如果观察到多个null，则所有null都将填充该会话的第一个非null。我需要last非null值来向前传播。

以下调整适用于我的用例：

def fill_forward(df, id_column, key_column, fill_column):

    # Fill null's with last *non null* value in the window
    ff = df.withColumn(
        'fill_fwd',
        func.last(fill_column, True) # True: fill with last non-null
        .over(
            Window.partitionBy(id_column)
            .orderBy(key_column)
            .rowsBetween(-sys.maxsize, 0))
        )

    # Drop the old column and rename the new column
    ff_out = ff.drop(fill_column).withColumnRenamed('fill_fwd', fill_column)

    return ff_out