PySpark DataFrame中基于前一个非空值顺序填充缺失数据

场景描述与问题分析

在数据处理过程中，我们经常会遇到需要根据序列中前一个有效值来填充后续缺失值的情况，这被称为“顺序填充”或“前向填充”（forward fill）。例如，在一个pyspark dataframe中，如果存在一个 row_id 字段表示数据的顺序，以及一个 group_id 字段，其中 group_id 仅在每个组的起始行有值，而后续行则为 null，直到下一个 group_id 出现。我们的目标是将这些 null 值填充为其所属组的第一个有效 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
...

这种场景在处理日志数据、时间序列数据或需要按逻辑分组填充的业务数据时非常常见。

解决方案：利用PySpark窗口函数实现顺序填充

PySpark的窗口函数（Window Functions）为处理此类序列依赖型问题提供了强大且高效的工具。通过定义一个合适的窗口，我们可以访问当前行之前（或之后）的数据，并应用聚合函数。

核心思路是：

硅基智能

基于Web3.0的元宇宙，去中心化的互联网，高质量、沉浸式元宇宙直播平台，用数字化重新定义直播

62

查看详情

1. 定义窗口： 创建一个基于 row_id 排序的窗口。
2. 应用聚合函数： 在这个窗口内，使用 last 函数并设置 ignorenulls=True 来获取当前行之前（包括当前行）的最后一个非空 group_id。

下面是具体的实现代码：

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

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

# 2. 准备示例数据
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. 定义窗口规范
# Window.orderBy("row_id") 确保数据按row_id升序处理
# rowsBetween(Window.unboundedPreceding, 0) 定义了从分区开始到当前行（包含当前行）的窗口范围
windowSpec = Window.orderBy("row_id").rowsBetween(Window.unboundedPreceding, 0)

# 4. 应用窗口函数进行缺失值填充
# F.last("group_id", ignorenulls=True) 获取窗口内最后一个非空值
filled_df = df.withColumn(
    "group_id",
    F.last("group_id", ignorenulls=True).over(windowSpec)
)

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

# 5. 停止SparkSession
spark.stop()

代码解释：

• SparkSession: Spark应用程序的入口点。
• data 和 columns: 用于创建示例DataFrame，模拟实际数据结构。
• Window.orderBy("row_id"): 这是定义窗口的关键部分，它指定了窗口内数据行的排序方式。对于顺序填充，必须按照 row_id（或任何表示序列的列）进行排序，以确保 last 函数能够正确地找到前一个有效值。
• rowsBetween(Window.unboundedPreceding, 0): 这定义了窗口的边界。
  • Window.unboundedPreceding 表示窗口从当前分区的第一行开始。
  • 0 表示窗口的结束点是当前行（currentRow 的别名）。
  • 结合起来，这个窗口包含了从分区开始到当前行的所有数据。
• F.last("group_id", ignorenulls=True).over(windowSpec): 这是应用窗口函数的核心。
  • F.last("group_id", ignorenulls=True): 这个聚合函数会返回指定列 group_id 在当前窗口中的最后一个值。ignorenulls=True 参数至关重要，它指示 last 函数在查找最后一个值时忽略 null 值，从而确保我们总是能找到最近的非空值。
  • .over(windowSpec): 将 last 函数应用于我们之前定义的 windowSpec 窗口。

注意事项与性能考量

1. row_id 的重要性: 确保 row_id 列是唯一且递增的，它决定了填充的顺序。如果 row_id 不唯一或顺序不正确，填充结果将不符合预期。
2. 窗口范围: rowsBetween(Window.unboundedPreceding, 0) 对于前向填充非常有效。如果需要其他类型的填充（例如后向填充或在特定组内填充），则需要相应调整窗口定义。
3. ignorenulls=True: 这是实现“基于前一个非空值填充”的关键。如果省略此参数或设置为 False，last 函数可能会返回 null，导致填充失败。
4. 大规模数据集性能: 窗口函数在PySpark中经过高度优化，能够高效处理大规模数据集（百万甚至数十亿行）。然而，窗口操作通常涉及数据的重分区和排序，这可能会消耗较多的计算资源。对于非常大的数据集，如果可能，可以考虑先对数据进行分区，以优化窗口操作的性能。
5. 替代方案对比:
   • fillna(): df.fillna(value) 只能用一个固定值或字典中的值填充所有 null，无法实现基于序列的动态填充。
   • UDF (User Defined Function): 虽然可以使用UDF实现复杂的填充逻辑，但UDF通常比内置函数和窗口函数效率低，尤其是在大规模数据上，不推荐用于此类场景。

总结

通过PySpark的窗口函数，特别是结合 Window.orderBy 和 F.last(ignorenulls=True)，我们可以优雅且高效地解决DataFrame中基于前一个非空值的顺序填充问题。这种方法不仅代码简洁，而且在处理大规模数据集时表现出良好的性能和可扩展性，是数据预处理中一项非常实用的技术。理解并熟练运用窗口函数，将大大提升PySpark数据处理的能力。

以上就是PySpark DataFrame中基于前一个非空值顺序填充缺失数据的详细内容，更多请关注php中文网其它相关文章！