視窗函式的定義引用一個大佬的定義： a window function calculates a return value for every input row of a table based on a group of rows。視窗函式與與其他函式的區別:

• 普通函式: 作用於每一條記錄，計算出一個新列(記錄數不變);
• 聚合函式: 作用於一組記錄(全部資料按照某種方式分為多組)，計算出一個聚合值(記錄數變小);
• 視窗函式: 作用於每一條記錄，逐條記錄去指定多條記錄來計算一個值(記錄數不變)。

視窗函式語法結構: 函式名(引數)OVER(PARTITION BY 子句 ORDER BY 子句 ROWS/RANGE子句)

• 函式名:
• OVER: 關鍵字，說明這是視窗函式，不是普通的聚合函式；
• 子句
  • PARTITION BY: 分組欄位
  • ORDER BY: 排序欄位
  • ROWS/RANG視窗子句: 用於控制視窗的尺寸邊界，有兩種(ROW,RANGE)
    • ROW: 物理視窗，資料篩選基於排序後的index
    • RANGE: 邏輯視窗，資料篩選基於值

主要有以下三種視窗函式

• ranking functions
• analytic functions
• aggregate functions

資料載入

from pyspark.sql.types import *


schema = StructType().add('name',StringType(),True).add('department',True).add('salary',IntegerType(),True)
df = spark.createDataFrame([
    ("Tom","Sales",4500),("Georgi",4200),("Kyoichi",3000),("Berni",4700),("Guoxiang",("Parto","Finance",2700),("Anneke",3300),("Sumant",3900),("Jeff","Marketing",3100),("Patricio",2500)
],schema=schema)
df.createOrReplaceTempView('salary')
df.show()

 

+--------+----------+------+
|    name|department|salary|
+--------+----------+------+
|     Tom|     Sales|  4500|
|  Georgi|     Sales|  4200|
| Kyoichi|     Sales|  3000|
|   Berni|     Sales|  4700|
|Guoxiang|     Sales|  4200|
|   Parto|   Finance|  2700|
|  Anneke|   Finance|  3300|
|  Sumant|   Finance|  3900|
|    Jeff| Marketing|  3100|
|Patricio| Marketing|  2500|
+--------+----------+------+
 

ranking functions

spark.sql("""
SELECT
    name,department,salary,row_number() over(partition by department order by salary) as index,rank() over(partition by department order by salary) as rank,dense_rank() over(partition by department order by salary) as dense_rank,percent_rank() over(partition by department order by salary) as percent_rank,ntile(2) over(partition by department order by salary) as ntile
FROM salary
""").toPandas()

analytic functions

spark.sql("""
SELECT
    name,cume_dist() over(partition by department order by salary) as cume_dist,lag('salary',2) over(partition by department order by salary) as lag,lead('salary',2) over(partition by department order by salary) as lead    
    
FROM salary
""").toPandas()

aggregate functions

只是在一定窗口裡實現一些普通的聚合函式。

spark.sql("""
SELECT
    name,sum(salary) over(partition by department order by salary) as sum,avg(salary) over(partition by department order by salary) as avg,min(salary) over(partition by department order by salary) as min,max(salary) over(partition by department order by salary) as max    
FROM salary
""").toPandas()

視窗子句

ROWS/RANG視窗子句: 用於控制視窗的尺寸邊界，有兩種(ROW,RANGE)

• ROWS: 物理視窗，資料篩選基於排序後的index
• RANGE: 邏輯視窗，資料篩選基於值

語法：OVER (PARTITION BY … ORDER BY … frame_type BETWEEN start AND end)

有以下5種邊界

• CURRENT ROW:
• UNBOUNDED PRECEDING: 分割槽第一行
• UNBOUNDED FOLLOWING: 分割槽最後一行
• n PRECEDING: 前n行
• n FOLLOWING: 後n行
• UNBOUNDED: 起點

spark.sql("""
SELECT
    name,row_number() over(partition by department order by salary rows between UNBOUNDED PRECEDING and CURRENT ROW) as index1
FROM salary
""").toPandas()

混合應用

spark.sql("""
SELECT
    name,salary - (min(salary) over(partition by department order by salary)) as salary_diff 
FROM salary
""").toPandas()

參考

• Introducing Window Functions in Spark SQL
• Standard Functions for Window Aggregation (Window Functions
• List Of Spark SQL Window Functions
• 在hive、Spark SQL中引入視窗函式