pyspark 笔记：窗口函数window-编程知识

窗口函数相关的概念和基本规范可以见：pyspark笔记：over-CSDN博客

1 创建Pyspark dataFrame

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

employee_salary = [("Ali", "Sales", 8000),("Bob", "Sales", 7000),("Cindy", "Sales", 7500),("Davd", "Finance", 10000),("Elena", "Sales", 8000),("Fancy", "Finance", 12000),("George", "Finance", 11000),("Haffman", "Marketing", 7000),("Ilaja", "Marketing", 8000),("Joey", "Sales", 9000)]columns= ["name", "department", "salary"]
df = spark.createDataFrame(data = employee_salary, schema = columns)
df.show(truncate=False)

2 定义窗口规范

以 partitionBy 作为分组条件，orderBy 对 Window 分组内的数据进行排序

# 以 department 字段进行分组，以 salary 倒序排序
# 按照部门对薪水排名，薪水最低的为第一名
windowSpec = Window.partitionBy("department").orderBy(F.asc("salary"))

后面的示例如无特殊说明，都是使用这个窗口规范

3 排名相关

3.1 `row_number()`

用于给出从1开始到每个窗口分区的结果的连续行号

df_part = df.withColumn("row_number", F.row_number().over(windowSpec)
)
df_part.show()

[观察上面的数据，发现同样的薪水会有不同的排名（比如Ali和Elena，都是8000的薪水，但一个第三一个第四），这是因为row_number()是按照行来给定序号，其不关注实际数值的大小。

3.1.1 应用举例：

找出每个department 薪水第二低的：

 df_part.where(F.col('row_number')==2).show()
'''
+------+----------+------+----------+
|  name|department|salary|row_number|
+------+----------+------+----------+
|George|   Finance| 11000|         2|
| Ilaja| Marketing|  8000|         2|
| Cindy|     Sales|  7500|         2|
+------+----------+------+----------+
'''

3.2 rank

不同于row_number，相同value的给相同值

df_part = df.withColumn('rank_number',F.rank().over(windowSpec))
df_part.show()
'''
+-------+----------+------+-----------+
|   name|department|salary|rank_number|
+-------+----------+------+-----------+
|   Davd|   Finance| 10000|          1|
| George|   Finance| 11000|          2|
|  Fancy|   Finance| 12000|          3|
|Haffman| Marketing|  7000|          1|
|  Ilaja| Marketing|  8000|          2|
|    Bob|     Sales|  7000|          1|
|  Cindy|     Sales|  7500|          2|
|    Ali|     Sales|  8000|          3|
|  Elena|     Sales|  8000|          3|
|   Joey|     Sales|  9000|          5|
+-------+----------+------+-----------+
'''

可以看到在rank下，Ali和Elena的rank_number是一样的了

3.3 dense rank

先看结果，再看和rank的区别

df_part = df.withColumn('dense_rank_number',F.dense_rank().over(windowSpec))
df_part.show()
+-------+----------+------+-----------------+
|   name|department|salary|dense_rank_number|
+-------+----------+------+-----------------+
|   Davd|   Finance| 10000|                1|
| George|   Finance| 11000|                2|
|  Fancy|   Finance| 12000|                3|
|Haffman| Marketing|  7000|                1|
|  Ilaja| Marketing|  8000|                2|
|    Bob|     Sales|  7000|                1|
|  Cindy|     Sales|  7500|                2|
|    Ali|     Sales|  8000|                3|
|  Elena|     Sales|  8000|                3|
|   Joey|     Sales|  9000|                4|
+-------+----------+------+-----------------+

我们重点看Joey（最后一行），rank中Ali和Elena并列第三后，身后的Joey排名第五；dense_rank中，Joey紧跟着排第四

3.4 percent_rank

百分位排名

df_part = df.withColumn('percent_rank_number',F.percent_rank().over(windowSpec))
df_part.show()
'''
+-------+----------+------+-------------------+
|   name|department|salary|percent_rank_number|
+-------+----------+------+-------------------+
|   Davd|   Finance| 10000|                0.0|
| George|   Finance| 11000|                0.5|
|  Fancy|   Finance| 12000|                1.0|
|Haffman| Marketing|  7000|                0.0|
|  Ilaja| Marketing|  8000|                1.0|
|    Bob|     Sales|  7000|                0.0|
|  Cindy|     Sales|  7500|               0.25|
|    Ali|     Sales|  8000|                0.5|
|  Elena|     Sales|  8000|                0.5|
|   Joey|     Sales|  9000|                1.0|
+-------+----------+------+-------------------+
'''

3.5 ntile

ntile()可将分组的数据按照指定数值n切分为n个部分，每一部分按照行的先后给定相同的序数。
例如n指定为2，则将组内数据分为两个部分，第一部分序号为1，第二部分序号为2。
理论上两部分数据行数是均等的，但当数据为奇数行时，中间的那一行归到前一部分。

df_part = df.withColumn('ntile(2)',F.ntile(2).over(windowSpec))
df_part.show()
'''
+-------+----------+------+--------+
|   name|department|salary|ntile(2)|
+-------+----------+------+--------+
|   Davd|   Finance| 10000|       1|
| George|   Finance| 11000|       1|
|  Fancy|   Finance| 12000|       2|
|Haffman| Marketing|  7000|       1|
|  Ilaja| Marketing|  8000|       2|
|    Bob|     Sales|  7000|       1|
|  Cindy|     Sales|  7500|       1|
|    Ali|     Sales|  8000|       1|
|  Elena|     Sales|  8000|       2|
|   Joey|     Sales|  9000|       2|
+-------+----------+------+--------+
'''

4 分析相关函数

4.1 `cume_dist`

数值的累进分布值

df.withColumn('cum_dist',F.cume_dist().over(windowSpec)).show()
'''
+-------+----------+------+------------------+
|   name|department|salary|          cum_dist|
+-------+----------+------+------------------+
|   Davd|   Finance| 10000|0.3333333333333333|
| George|   Finance| 11000|0.6666666666666666|
|  Fancy|   Finance| 12000|               1.0|
|Haffman| Marketing|  7000|               0.5|
|  Ilaja| Marketing|  8000|               1.0|
|    Bob|     Sales|  7000|               0.2|
|  Cindy|     Sales|  7500|               0.4|
|    Ali|     Sales|  8000|               0.8|
|  Elena|     Sales|  8000|               0.8|
|   Joey|     Sales|  9000|               1.0|
+-------+----------+------+------------------+
'''

这个表怎么解读呢？以Sales为例，薪资小于等于7000的占比0.2，薪资小于等于7500的占比0.4，以此类推

4.2 lag

照指定列排好序的分组内每个数值的上一个数值

df.withColumn('lag',F.lag('name').over(windowSpec)).show()
'''
+-------+----------+------+-------+
|   name|department|salary|    lag|
+-------+----------+------+-------+
|   Davd|   Finance| 10000|   null|
| George|   Finance| 11000|   Davd|
|  Fancy|   Finance| 12000| George|
|Haffman| Marketing|  7000|   null|
|  Ilaja| Marketing|  8000|Haffman|
|    Bob|     Sales|  7000|   null|
|  Cindy|     Sales|  7500|    Bob|
|    Ali|     Sales|  8000|  Cindy|
|  Elena|     Sales|  8000|    Ali|
|   Joey|     Sales|  9000|  Elena|
+-------+----------+------+-------+
'''

4.3 lead

和lag相反，下一个值

df.withColumn('lead',F.lead('name').over(windowSpec)).show()
'''
+-------+----------+------+------+
|   name|department|salary|  lead|
+-------+----------+------+------+
|   Davd|   Finance| 10000|George|
| George|   Finance| 11000| Fancy|
|  Fancy|   Finance| 12000|  null|
|Haffman| Marketing|  7000| Ilaja|
|  Ilaja| Marketing|  8000|  null|
|    Bob|     Sales|  7000| Cindy|
|  Cindy|     Sales|  7500|   Ali|
|    Ali|     Sales|  8000| Elena|
|  Elena|     Sales|  8000|  Joey|
|   Joey|     Sales|  9000|  null|
+-------+----------+------+------+
'''

5 聚合函数

此时的聚合样式为：

windowSpecAgg=Window.partitionBy('department')

5.1 avg

平均值

df.withColumn('avg',F.avg('salary').over(windowSpecAgg)).show()
'''
+-------+----------+------+-------+
|   name|department|salary|    avg|
+-------+----------+------+-------+
|   Davd|   Finance| 10000|11000.0|
|  Fancy|   Finance| 12000|11000.0|
| George|   Finance| 11000|11000.0|
|Haffman| Marketing|  7000| 7500.0|
|  Ilaja| Marketing|  8000| 7500.0|
|    Ali|     Sales|  8000| 7900.0|
|    Bob|     Sales|  7000| 7900.0|
|  Cindy|     Sales|  7500| 7900.0|
|  Elena|     Sales|  8000| 7900.0|
|   Joey|     Sales|  9000| 7900.0|
+-------+----------+------+-------+
'''

5.2 sum 求和

5.3 min/max 最大最小值

5.4 count 这一个窗口内有多少记录

df.withColumn('count',F.count('salary').over(windowSpecAgg)).show()
'''
+-------+----------+------+-----+
|   name|department|salary|count|
+-------+----------+------+-----+
|   Davd|   Finance| 10000|    3|
|  Fancy|   Finance| 12000|    3|
| George|   Finance| 11000|    3|
|Haffman| Marketing|  7000|    2|
|  Ilaja| Marketing|  8000|    2|
|    Ali|     Sales|  8000|    5|
|    Bob|     Sales|  7000|    5|
|  Cindy|     Sales|  7500|    5|
|  Elena|     Sales|  8000|    5|
|   Joey|     Sales|  9000|    5|
+-------+----------+------+-----+
'''

5.5 approx_count_distinct 相同的值只记录一次

df.withColumn('ap_count',F.approx_count_distinct('salary').over(windowSpecAgg)).show()
'''
+-------+----------+------+--------+
|   name|department|salary|ap_count|
+-------+----------+------+--------+
|   Davd|   Finance| 10000|       3|
|  Fancy|   Finance| 12000|       3|
| George|   Finance| 11000|       3|
|Haffman| Marketing|  7000|       2|
|  Ilaja| Marketing|  8000|       2|
|    Ali|     Sales|  8000|       4|
|    Bob|     Sales|  7000|       4|
|  Cindy|     Sales|  7500|       4|
|  Elena|     Sales|  8000|       4|
|   Joey|     Sales|  9000|       4|
+-------+----------+------+--------+
'''

参考内容：【PySpark】窗口函数Window - 知乎