PySpark中遇item_id为空时重新分配会话ID的实现方法

实现思路 #

核心是通过窗口函数统计每个会话内当前行之前出现的item_id空值次数，以此作为分组依据；再为每个分组获取起始pos值，拼接生成新会话ID；最后处理空值行的会话ID为None。

具体代码实现 #

1. 导入必要的函数和窗口模块：

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

2. 定义基础窗口（按会话ID分区，按位置排序）：

w = Window.partitionBy("session_id").orderBy("pos")

3. 统计当前行之前的空值次数，用于划分会话分组：

df_with_group = df1.withColumn(
    "null_count_before",
    F.sum(F.when(F.col("item_id").isNull(), 1).otherwise(0)).over(w.rowsBetween(Window.unboundedPreceding, -1))
).fillna({"null_count_before": 0})  # 处理第一行无前置数据的情况

4. 为每个分组计算起始pos值：

w_group = Window.partitionBy("session_id", "null_count_before")
df_with_start_pos = df_with_group.withColumn(
    "start_pos",
    F.min(F.col("pos")).over(w_group)
)

5. 生成新会话ID并清理临时列：

result_df = df_with_start_pos.withColumn(
    "new_session_id",
    F.when(
        F.col("item_id").isNull(),
        F.lit(None)
    ).otherwise(
        F.concat(F.col("session_id"), F.lit("_"), F.col("start_pos").cast("string"))
    )
).drop("null_count_before", "start_pos")

6. 查看结果：

result_df.show(truncate=False)

结果说明 #

执行上述代码后，会得到与预期一致的输出：

• 空值行的new_session_id为None
• 空值之前的非空项归为session_id_起始pos（如s1_0）
• 空值之后的非空项归为新的会话（如s1_4）

内容的提问来源于stack exchange，提问作者n0obcoder