PySpark填充缺失值方法详解

本文介绍了如何使用 PySpark 顺序填充 DataFrame 中的缺失值。通过使用窗口函数和 last 函数，我们可以高效地将每个 group_id 中的空值填充为该组的第一个非空值，从而解决在大型 DataFrame 中处理缺失值的问题。该方法适用于已知 row_id 是顺序且唯一的情况。

在数据处理过程中，经常会遇到需要填充 DataFrame 中缺失值的情况。当缺失值具有一定的顺序性或分组性时，例如，希望将每个分组中的缺失值填充为该分组的第一个有效值，可以使用 PySpark 的窗口函数来实现。以下将详细介绍如何使用 PySpark 顺序填充 DataFrame 中的缺失值。

准备工作

首先，需要创建一个 SparkSession 并准备好示例数据。

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

# 创建 SparkSession
spark = SparkSession.builder.appName("FillMissingValues").getOrCreate()

# 示例数据
data = [(1, 1), (2, None), (3, None), (4, None), (5, 5), (6, None), (7, None), (8, 8), (9, None), (10, None), (11, None), (12, None)]
columns = ["row_id", "group_id"]
df = spark.createDataFrame(data, columns)

df.show()

这段代码首先导入必要的库，包括 SparkSession 用于创建 Spark 会话，functions (别名 F) 用于使用 Spark 内置函数，以及 Window 用于定义窗口规范。然后，创建一个名为 "FillMissingValues" 的 SparkSession。接着，创建包含 row_id 和 group_id 的示例数据，并将其转换为 PySpark DataFrame。最后，使用 df.show() 显示 DataFrame 的内容。

使用窗口函数填充缺失值

核心思路是使用窗口函数 last 来查找每个 row_id 之前（包括当前行）的最后一个非空 group_id 值。通过定义一个基于 row_id 排序的窗口，并设置其范围为从第一行到当前行，就可以实现顺序填充的效果。

# 定义窗口规范
windowSpec = Window.orderBy("row_id").rowsBetween(Window.unboundedPreceding, 0)

# 使用 last 函数填充缺失值
filled_df = df.withColumn("group_id", F.last("group_id", ignorenulls=True).over(windowSpec))

# 显示结果
filled_df.show()

这段代码首先定义了一个窗口规范 windowSpec，该规范基于 row_id 进行排序，并指定窗口的范围为从第一行 (Window.unboundedPreceding) 到当前行 (0)。然后，使用 withColumn 函数创建一个新的 group_id 列，其值为 F.last("group_id", ignorenulls=True).over(windowSpec)。F.last 函数查找窗口中最后一个非空 (ignorenulls=True) 的 group_id 值，并将其应用于每个 row_id。最后，使用 filled_df.show() 显示填充后的 DataFrame。

代码解释

• Window.orderBy("row_id"): 定义窗口的排序方式，这里按照 row_id 升序排列。确保了顺序填充的正确性。
• Window.rowsBetween(Window.unboundedPreceding, 0): 定义窗口的范围。Window.unboundedPreceding 表示窗口从第一行开始，0 表示窗口到当前行结束。这意味着对于每一行，窗口都包含从第一行到当前行的所有行。
• F.last("group_id", ignorenulls=True).over(windowSpec): F.last 函数用于获取窗口中最后一个非空值。ignorenulls=True 参数表示忽略空值，只查找非空值。over(windowSpec) 将 last 函数应用到定义的窗口上。

注意事项

• 确保 row_id 是顺序且唯一的。如果 row_id 不是顺序的，则需要先对其进行排序。
• 如果数据量非常大，可以考虑对 DataFrame 进行分区，以提高计算效率。
• 此方法适用于将缺失值填充为该组的第一个有效值。如果需要使用其他填充策略，例如使用平均值或中位数，则需要使用不同的函数。

总结

通过使用 PySpark 的窗口函数和 last 函数，可以方便地实现 DataFrame 中缺失值的顺序填充。这种方法在处理具有分组性和顺序性的数据时非常有效，可以帮助我们更好地进行数据清洗和分析。