PySpark中利用窗口函数按序填充DataFrame缺失值的高效策略

1. 理解问题背景与需求

在数据处理过程中，我们经常会遇到dataframe中存在序列性缺失值的情况。例如，在一个包含 row_id 和 group_id 的pyspark dataframe中，row_id 是一个递增且唯一的序列号，而 group_id 则表示一个组的唯一标识。当 group_id 首次出现一个非空值时，它标志着一个新组的开始，此后的 null 值都应填充为该组的起始 group_id，直到下一个非空 group_id 出现。

例如，原始数据可能如下：

row_id, group_id
1,      1
2,      null
3,      null
4,      null
5,      5
6,      null
7,      null
8,      8
...

我们的目标是将其转换为：

row_id, group_id
1,      1
2,      1
3,      1
4,      1
5,      5
6,      5
7,      5
8,      8
...

这种填充需求在大规模数据集（例如数百万甚至数十亿条记录）上需要高效的解决方案。

2. PySpark窗口函数概述

PySpark的窗口函数提供了一种强大的机制，允许我们在DataFrame的特定“窗口”内执行计算。一个窗口定义了一组与当前行相关的行，并且可以根据一个或多个列进行排序。在处理序列性数据和聚合操作时，窗口函数表现出卓越的灵活性和性能。

SpeakingPass-打造你的专属雅思口语语料

使用chatGPT帮你快速备考雅思口语，提升分数

25

查看详情

本教程将利用以下窗口函数特性：

• Window.orderBy(): 定义窗口内行的排序顺序，这对于序列性填充至关重要。
• rowsBetween(): 进一步限定窗口的范围，例如从窗口的起始到当前行。
• F.last(): 获取窗口内指定列的最后一个值。
• ignorenulls=True: 在 last() 函数中，忽略 null 值，只考虑非 null 值。

3. 核心解决方案：使用 last 函数与窗口规范

解决此问题的关键在于正确定义窗口规范，并利用 last 函数在窗口内获取最近的非空 group_id。

3.1 步骤详解

1. 创建SparkSession: 初始化Spark环境。
2. 准备DataFrame: 构建一个示例DataFrame，模拟实际数据结构。
3. 定义窗口规范:
   • 使用 Window.orderBy("row_id") 确保窗口内的行按照 row_id 升序排列，这是实现序列性填充的基础。
   • 使用 rowsBetween(Window.unboundedPreceding, 0) 定义窗口范围。这意味着对于当前行，窗口将包括从分区开始到当前行（包括当前行）的所有行。Window.unboundedPreceding 表示窗口的起始点是分区的第一行，0 表示窗口的结束点是当前行。
4. 应用 last 函数:
   • F.last("group_id", ignorenulls=True)：这个函数将在我们定义的窗口内查找 group_id 列的最后一个非 null 值。ignorenulls=True 参数是至关重要的，它确保我们只考虑非空的 group_id 值进行填充。
   • .over(windowSpec)：将 last 函数应用到之前定义的 windowSpec 窗口上。

3.2 示例代码

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

# 1. 创建一个SparkSession
spark = SparkSession.builder.appName("SequentialNullFill").getOrCreate()

# 2. 准备示例DataFrame
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)

print("原始DataFrame:")
df.show()

# 3. 定义窗口规范
# 窗口按 row_id 升序排列
# 范围是从分区开始到当前行（包括当前行）
windowSpec = Window.orderBy("row_id").rowsBetween(Window.unboundedPreceding, 0)

# 4. 使用 last 窗口函数填充 null 值
# ignorenulls=True 确保只考虑非空值
filled_df = df.withColumn(
    "group_id",
    F.last("group_id", ignorenulls=True).over(windowSpec)
)

print("填充缺失值后的DataFrame:")
filled_df.show()

# 停止SparkSession
spark.stop()

3.3 代码执行结果

原始DataFrame:
+------+--------+
|row_id|group_id|
+------+--------+
|     1|       1|
|     2|    null|
|     3|    null|
|     4|    null|
|     5|       5|
|     6|    null|
|     7|    null|
|     8|       8|
|     9|    null|
|    10|    null|
|    11|    null|
|    12|    null|
+------+--------+

填充缺失值后的DataFrame:
+------+--------+
|row_id|group_id|
+------+--------+
|     1|       1|
|     2|       1|
|     3|       1|
|     4|       1|
|     5|       5|
|     6|       5|
|     7|       5|
|     8|       8|
|     9|       8|
|    10|       8|
|    11|       8|
|    12|       8|
+------+--------+

4. 注意事项与性能考量

1. row_id 的重要性: 此方法依赖于 row_id 的递增和唯一性来正确地定义序列顺序。如果 row_id 不具备这些特性，需要先对其进行预处理或选择其他合适的排序键。
2. 窗口范围: rowsBetween(Window.unboundedPreceding, 0) 是此解决方案的核心。它确保了在计算当前行的 group_id 时，只考虑了当前行及之前的所有行中的非空 group_id。如果使用 Window.unboundedFollowing 或其他范围，结果可能会不符合预期。
3. 性能: 对于大规模数据集，窗口函数通常比UDF（用户自定义函数）或迭代操作更高效，因为它们可以在Spark的优化器中进行优化。然而，Window.unboundedPreceding 意味着每个任务可能需要处理大量数据，这在极端情况下可能导致内存压力。如果DataFrame非常庞大且分区数不足，可能会影响性能。适当的分区策略（例如，如果存在更高级别的分组，可以在 Window.partitionBy() 中指定）可以进一步优化性能。
4. ignorenulls=True: 务必包含此参数，否则 last 函数可能会返回 null 值，如果窗口的最后一个值恰好是 null。
5. 数据类型: 确保 group_id 列的数据类型能够支持填充后的值。

5. 总结

本教程详细阐述了如何在PySpark DataFrame中，利用窗口函数 (Window) 结合 last 函数和 ignorenulls=True 参数，高效地实现序列性缺失值填充。通过定义正确的窗口规范 (Window.orderBy("row_id").rowsBetween(Window.unboundedPreceding, 0))，我们能够将 group_id 列中的 null 值填充为其所在序列中最近的非空值，从而满足数据连续性的需求。此方法在处理大规模数据集时表现出良好的性能和扩展性，是PySpark数据清洗和预处理中的一个重要技巧。

以上就是PySpark中利用窗口函数按序填充DataFrame缺失值的高效策略的详细内容，更多请关注php中文网其它相关文章！