pyspark 窗口函数

PySpark中的窗口函数是一种用于在数据集上进行聚合、排序和分析的强大工具。它们允许你在数据集的特定子集上执行聚合操作，而不需要将整个数据集加载到内存中。

在PySpark中使用窗口函数，你需要首先导入相关的模块：

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, rank, dense_rank, percent_rank, first, last, lag, lead

然后，你可以使用`Window.partitionBy()`方法指定一个或多个分区列，以确定窗口函数的作用范围。例如，假设你有一个名为`df`的数据集，包含`id`、`category`和`value`这三列，你可以按照`category`进行分区，并按照`value`进行排序：

windowSpec = Window.partitionBy("category").orderBy("value")

接下来，你可以使用各种窗口函数对数据进行处理。以下是几个常用的窗口函数示例：

- `row_number()`：为每个分区中的行分配唯一的序号。
- `rank()`：计算每个分区中行的排名（相同值得到相同的排名，下一个排名将被跳过）。
- `dense_rank()`：计算每个分区中行的密集排名（相同值得到相同的排名，下一个排名不会被跳过）。
- `percent_rank()`：计算每个分区中行的百分比排名。
- `first()`：返回每个分区中的第一行。
- `last()`：返回每个分区中的最后一行。
- `lag()`：返回当前行之前的指定偏移量的行。
- `lead()`：返回当前行之后的指定偏移量的行。

你可以使用这些窗口函数来创建新的列，例如：

from pyspark.sql.functions import sum, avg

df.withColumn("row_number", row_number().over(windowSpec)) \
  .withColumn("rank", rank().over(windowSpec)) \
  .withColumn("dense_rank", dense_rank().over(windowSpec)) \
  .withColumn("percent_rank", percent_rank().over(windowSpec)) \
  .withColumn("sum_value", sum("value").over(windowSpec)) \
  .withColumn("avg_value", avg("value").over(windowSpec)) \
  .show()

这将在原始数据集的基础上添加了几个新的列，并应用了各种窗口函数。

希望这能帮助到你！如果你有任何其他问题，请随时问我。