窗口函數(shù)對數(shù)據(jù)組進(jìn)行操作，并為每個記錄或組返回值

 

[[331416]]
> Photo by Tom Blackout on Unsplash

在此博客文章中，我們將深入探討Apache Spark窗口函數(shù)。 您可能也對我之前有關(guān)Apache Spark的帖子感興趣。

• 使用Apache Spark開始您的旅程-第1部分
• 使用Apache Spark開始您的旅程-第2部分
• Apache Spark開始您的旅程-第3部分
• 深入研究Apache Spark DateTime函數(shù)
• 在Apache Spark中使用JSON

首先，讓我們看看什么是窗口函數(shù)以及何時使用它們。 我們在Apache Spark中使用了各種功能，例如月份(從日期返回月份)，四舍五入(舍入值)和地板(為給定的輸入提供底值)等，這些功能將在每條記錄上執(zhí)行并返回一個值 每條記錄。 然后，我們將對一組數(shù)據(jù)執(zhí)行各種聚合函數(shù)，并為每個組返回一個值，例如sum，avg，min，max和count。 但是，如果我們想對一組數(shù)據(jù)執(zhí)行該操作，并且希望對每個記錄有一個單一的值/結(jié)果怎么辦? 在這種情況下，我們可以使用窗口函數(shù)。 他們可以定義記錄的排名，累積分布，移動平均值，或標(biāo)識當(dāng)前記錄之前或之后的記錄。

讓我們使用一些Scala API示例來了解以下窗口函數(shù)：

• 匯總：min, max, avg, count, 和 sum.
• 排名：rank，dense_rank，percent_rank，row_num和ntile
• 分析性：cume_dist，lag和lead
• 自定義邊界：rangeBetween和rowsBetween

為便于參考，GitHub上提供了一個以JSON文件格式導(dǎo)出的Zeppelin筆記本和一個Scala文件。

創(chuàng)建Spark DataFrame

現(xiàn)在，我們創(chuàng)建一個示例Spark DataFrame，我們將在整個博客中使用它。 首先，讓我們加載所需的庫。

import org.apache.spark.sql.expressions.Window 
import org.apache.spark.sql.functions._ 

現(xiàn)在，我們將使用一些虛擬數(shù)據(jù)創(chuàng)建DataFrame，這些虛擬數(shù)據(jù)將用于討論各種窗口函數(shù)。

case class Salary(depName: String, empNo: Long, salary: Long) 
 
val empsalary = Seq( 
  Salary("sales", 1, 5000), 
  Salary("personnel", 2, 3900), 
  Salary("sales", 3, 4800), 
  Salary("sales", 4, 4800), 
  Salary("personnel", 5, 3500), 
  Salary("develop", 7, 4200), 
  Salary("develop", 8, 6000), 
  Salary("develop", 9, 4500), 
  Salary("develop", 10, 5200), 
  Salary("develop", 11, 5200)).toDF() 

這是我們的DataFrame的樣子：

 

 

窗口集合函數(shù)

讓我們看一些聚合的窗口函數(shù)，看看它們?nèi)绾喂ぷ鳌?/p>

首先，我們需要定義窗口的規(guī)范。 假設(shè)我們要根據(jù)部門獲取匯總數(shù)據(jù)。 因此，在此示例中，我們將基于部門名稱(列：depname)定義窗口。

為聚合函數(shù)創(chuàng)建窗口規(guī)范

val byDepName = Window.partitionBy("depName") 

在窗口上應(yīng)用聚合函數(shù)

現(xiàn)在，在部門內(nèi)(列：depname)，我們可以應(yīng)用各種聚合函數(shù)。 因此，讓我們嘗試查找每個部門的最高和最低工資。 在這里，我們僅選擇了所需的列(depName，max_salary和min_salary)，并刪除了重復(fù)的記錄。

val agg_sal = empsalary 
           .withColumn("max_salary", max("salary").over(byDepName)) 
           .withColumn("min_salary", min("salary").over(byDepName)) 
                 
 
agg_sal.select("depname", "max_salary", "min_salary") 
        .dropDuplicates() 
        .show() 

輸出：

+---------+----------+----------+ 
|  depname|max_salary|min_salary| 
+---------+----------+----------+ 
|  develop|      6000|      4200| 
|    sales|      5000|      4800| 
|personnel|      3900|      3500| 
+---------+----------+----------+ 

現(xiàn)在讓我們看看它是如何工作的。 我們已經(jīng)按部門名稱對數(shù)據(jù)進(jìn)行了分區(qū)：

 

現(xiàn)在，當(dāng)我們執(zhí)行合計(jì)函數(shù)時，它將應(yīng)用于每個分區(qū)并返回合計(jì)值(在本例中為min和max)。

 

注意：可用的匯總函數(shù)為最大，最小，總和，平均和計(jì)數(shù)。

窗口排名功能

在本節(jié)中，我們將討論幾種類型的排名函數(shù)。

創(chuàng)建用于排序功能的窗口規(guī)范

現(xiàn)在，我們要根據(jù)員工在部門內(nèi)的薪水進(jìn)行排名。 薪水最高的員工將排名第一，薪水最低的員工將排名最后。 在這里，我們將基于部門(列：depname)對數(shù)據(jù)進(jìn)行分區(qū)，并且在部門內(nèi)，我們將根據(jù)薪水以降序?qū)?shù)據(jù)進(jìn)行排序。

val winSpec = Window.partitionBy("depName").orderBy("salary".desc) 

對于每個部門，記錄將根據(jù)薪水以降序排序。

 

1.等級功能：等級

此函數(shù)將返回分區(qū)中每個記錄的等級，并跳過任何重復(fù)等級之后的后續(xù)等級：

val rank_df = empsalary.withColumn("rank", rank().over(winSpec)) 
rank_df.show() 

輸出：

+---------+-----+------+----+ 
|  depName|empNo|salary|rank| 
+---------+-----+------+----+ 
|  develop|    8|  6000|   1| 
|  develop|   11|  5200|   2| 
|  develop|   10|  5200|   2| 
|  develop|    9|  4500|   4| 
|  develop|    7|  4200|   5| 
|    sales|    1|  5000|   1| 
|    sales|    4|  4800|   2| 
|    sales|    3|  4800|   2| 
|personnel|    2|  3900|   1| 
|personnel|    5|  3500|   2| 
+---------+-----+------+----+ 

在這里我們可以看到某些等級重復(fù)，而有些等級丟失。 例如，在開發(fā)部門中，我們有2名等級= 2的員工，而沒有等級= 3的員工，因?yàn)榈燃壓瘮?shù)將為相同的值保留相同的等級，并相應(yīng)地跳過下一個等級。

2.密集等級：densed_rank

此函數(shù)將返回分區(qū)中每個記錄的等級，但不會跳過任何等級。

val dense_rank_df = empsalary.withColumn("dense_rank", dense_rank().over(winSpec)) 
 
dense_rank_df.show() 

輸出：

+---------+-----+------+-----------+ 
|  depName|empNo|salary|desnse_rank| 
+---------+-----+------+-----------+ 
|  develop|    8|  6000|          1| 
|  develop|   10|  5200|          2| 
|  develop|   11|  5200|          2| 
|  develop|    9|  4500|          3| 
|  develop|    7|  4200|          4| 
|    sales|    1|  5000|          1| 
|    sales|    3|  4800|          2| 
|    sales|    4|  4800|          2| 
|personnel|    2|  3900|          1| 
|personnel|    5|  3500|          2| 
+---------+-----+------+-----------+ 

在這里，我們可以看到某些等級是重復(fù)的，但是排名并沒有像我們使用等級功能時那樣丟失。 例如，在開發(fā)部門中，我們有2名員工，等級=2。density_rank函數(shù)將為相同的值保留相同的等級，但不會跳過下一個等級。

3.行號函數(shù)：row_number

此功能將在窗口內(nèi)分配行號。 如果2行的排序列值相同，則不確定將哪個行號分配給具有相同值的每一行。

val row_num_df = empsalary.withColumn("row_number", row_number().over(winSpec)) 
 
row_num_df.show() 

輸出：

+---------+-----+------+----------+ 
|  depName|empNo|salary|row_number| 
+---------+-----+------+----------+ 
|  develop|    8|  6000|         1| 
|  develop|   10|  5200|         2| 
|  develop|   11|  5200|         3| 
|  develop|    9|  4500|         4| 
|  develop|    7|  4200|         5| 
|    sales|    1|  5000|         1| 
|    sales|    3|  4800|         2| 
|    sales|    4|  4800|         3| 
|personnel|    2|  3900|         1| 
|personnel|    5|  3500|         2| 
+---------+-----+------+----------+ 

4.百分比排名函數(shù)：percent_rank

此函數(shù)將返回分區(qū)中的相對(百分?jǐn)?shù))等級。

val percent_rank_df = empsalary.withColumn("percent_rank", percent_rank().over(winSpec)) 
 
percent_rank_df.show() 

輸出：

+---------+-----+------+------------+ 
|  depName|empNo|salary|percent_rank| 
+---------+-----+------+------------+ 
|  develop|    8|  6000|         0.0| 
|  develop|   10|  5200|        0.25| 
|  develop|   11|  5200|        0.25| 
|  develop|    9|  4500|        0.75| 
|  develop|    7|  4200|         1.0| 
|    sales|    1|  5000|         0.0| 
|    sales|    3|  4800|         0.5| 
|    sales|    4|  4800|         0.5| 
|personnel|    2|  3900|         0.0| 
|personnel|    5|  3500|         1.0| 
+---------+-----+------+------------+ 

5. N-tile功能：ntile

此功能可以根據(jù)窗口規(guī)格或分區(qū)將窗口進(jìn)一步細(xì)分為n個組。 例如，如果需要將部門進(jìn)一步劃分為三類，則可以將ntile指定為3。

val ntile_df = empsalary.withColumn("ntile", ntile(3).over(winSpec)) 
 
ntile_df.show() 

輸出：

+---------+-----+------+-----+ 
|  depName|empNo|salary|ntile| 
+---------+-----+------+-----+ 
|  develop|    8|  6000|    1| 
|  develop|   10|  5200|    1| 
|  develop|   11|  5200|    2| 
|  develop|    9|  4500|    2| 
|  develop|    7|  4200|    3| 
|    sales|    1|  5000|    1| 
|    sales|    3|  4800|    2| 
|    sales|    4|  4800|    3| 
|personnel|    2|  3900|    1| 
|personnel|    5|  3500|    2| 
+---------+-----+------+-----+ 

 

窗口分析功能

接下來，我們將討論諸如累積分布，滯后和超前的分析功能。

1.累積分布函數(shù)：cume_dist

此函數(shù)提供窗口/分區(qū)的值的累積分布。

val winSpec = Window.partitionBy("depName").orderBy("salary") 
 
val cume_dist_df =  
              empsalary.withColumn("cume_dist",cume_dist().over(winSpec)) 
 
cume_dist_df.show() 

定義窗口規(guī)范并應(yīng)用cume_dist函數(shù)以獲取累積分布。

輸出：

+---------+-----+------+------------------+ 
|  depName|empNo|salary|         cume_dist| 
+---------+-----+------+------------------+ 
|  develop|    7|  4200|               0.2| 
|  develop|    9|  4500|               0.4| 
|  develop|   10|  5200|               0.8| 
|  develop|   11|  5200|               0.8| 
|  develop|    8|  6000|               1.0| 
|    sales|    4|  4800|0.6666666666666666| 
|    sales|    3|  4800|0.6666666666666666| 
|    sales|    1|  5000|               1.0| 
|personnel|    5|  3500|               0.5 
|personnel|    2|  3900|               1.0| 
+---------+-----+------+------------------+ 

2.滯后功能：滯后

此函數(shù)將在從DataFrame偏移行之前返回該值。

lag函數(shù)采用3個參數(shù)(lag(col，count = 1，默認(rèn)= None))，col：定義需要在其上應(yīng)用函數(shù)的列。 count：需要回顧多少行。 default：定義默認(rèn)值。

val winSpec = Window.partitionBy("depName").orderBy("salary") 
 
val lag_df =  
          empsalary.withColumn("lag", lag("salary", 2).over(winSpec)) 
 
lag_df.show() 

輸出：

+---------+-----+------+----+ 
|  depName|empNo|salary| lag| 
+---------+-----+------+----+ 
|  develop|    7|  4200|null| 
|  develop|    9|  4500|null| 
|  develop|   10|  5200|4200| 
|  develop|   11|  5200|4500| 
|  develop|    8|  6000|5200| 
|    sales|    4|  4800|null| 
|    sales|    3|  4800|null| 
|    sales|    1|  5000|4800| 
|personnel|    5|  3500|null| 
|personnel|    2|  3900|null| 
+---------+-----+------+----+ 

例如，讓我們在當(dāng)前行之前查找薪水2行。

• 對于depname = develop，salary = 4500。沒有這樣的行，該行在該行之前2行。 因此它將為空。

 

• 對于部門名稱=發(fā)展，薪水= 6000(以藍(lán)色突出顯示)。 如果我們提前兩排，我們將獲得5200的薪水(以綠色突出顯示)。

 

3.導(dǎo)聯(lián)功能：導(dǎo)聯(lián)

此函數(shù)將返回DataFrame的偏移行之后的值。

val winSpec = Window.partitionBy("depName").orderBy("salary") 
 
val lead_df =  
          empsalary.withColumn("lead", lead("salary", 2).over(winSpec)) 
 
lead_df.show() 

lead函數(shù)采用3個參數(shù)(lead(col，count = 1，默認(rèn)= None))col：定義需要在其上應(yīng)用函數(shù)的列。 count：對于當(dāng)前行，我們需要向前/向后查找多少行。 default：定義默認(rèn)值。

輸出：

+---------+-----+------+----+ 
|  depName|empNo|salary| lag| 
+---------+-----+------+----+ 
|  develop|    7|  4200|5200| 
|  develop|    9|  4500|5200| 
|  develop|   10|  5200|6000| 
|  develop|   11|  5200|null| 
|  develop|    8|  6000|null| 
|    sales|    3|  4800|5000| 
|    sales|    4|  4800|null| 
|    sales|    1|  5000|null| 
|personnel|    5|  3500|null| 
|personnel|    2|  3900|null| 
+---------+-----+------+----+ 

讓我們嘗試從當(dāng)前行中查找前進(jìn)/后兩行的薪水。

• 對于depname =開發(fā)人員，薪水= 4500(以藍(lán)色突出顯示)。 如果我們在前進(jìn)/后退兩行，我們將獲得5200的薪水(以綠色突出顯示)。

 

• 對于depname =人員，薪水=3500。在此分區(qū)中，沒有此行向前2行/在該行之后。 因此我們將獲得空值。

 

 

自定義窗口定義

默認(rèn)情況下，窗口的邊界由分區(qū)列定義，我們可以通過窗口規(guī)范指定順序。 例如，對于開發(fā)部門，窗口的開始是工資的最小值，窗口的結(jié)束是工資的最大值。

但是，如果我們想更改窗口的邊界怎么辦? 以下功能可用于定義每個分區(qū)內(nèi)的窗口。

1. rangeBetween

使用rangeBetween函數(shù)，我們可以顯式定義邊界。例如，從當(dāng)前薪水開始，將其定義為100，然后將其定義為300，并查看其含義。 從100開始表示窗口將從100個單位開始，從當(dāng)前值開始以300個值結(jié)束(包括開始值和結(jié)束值)。

val winSpec = Window.partitionBy("depName") 
          .orderBy("salary") 
          .rangeBetween(100L, 300L) 

定義窗口規(guī)格

起始值和結(jié)束值后的L表示該值是Scala Long類型。

val range_between_df = empsalary.withColumn("max_salary", max("salary").over(winSpec)) 
 
range_between_df.show() 

應(yīng)用自定義窗口規(guī)范

輸出：

+---------+-----+------+----------+ 
|  depName|empNo|salary|max_salary| 
+---------+-----+------+----------+ 
|  develop|    7|  4200|      4500| 
|  develop|    9|  4500|      null| 
|  develop|   10|  5200|      null| 
|  develop|   11|  5200|      null| 
|  develop|    8|  6000|      null| 
|    sales|    3|  4800|      5000| 
|    sales|    4|  4800|      5000| 
|    sales|    1|  5000|      null| 
|personnel|    5|  3500|      null| 
|personnel|    2|  3900|      null| 
+---------+-----+------+----------+ 

現(xiàn)在，讓我們嘗試了解輸出。

• 對于depname = developer，salary = 4200，窗口的開始將是(當(dāng)前值+開始)，即4200 + 100 =4300。窗口的結(jié)束將是(當(dāng)前值+結(jié)束)，即4200 + 300 = 4500。

由于只有一個薪水值在4300到4500之間，包括開發(fā)部門的4500，所以我們將4500作為max_salary作為4200(上面的檢查輸出)。

 

• 同樣，對于depname = develop，salary = 4500，窗口將為(開始：4500 + 100 = 4600，結(jié)束：4500 + 300 = 4800)。 但是開發(fā)部門沒有薪水值在4600到4800之間，因此最大值不會為空(上面的檢查輸出)。

 

這里有一些特殊的邊界值可以使用。

• Window.currentRow：指定一行中的當(dāng)前值。
• Window.unboundedPreceding：這可以用于使窗口無限制地開始。
• Window.unbounded以下：此方法可用于使窗口具有無限的末端。

例如，我們需要從員工工資中找到最高工資，該最高工資大于300。 因此，我們將起始值定義為300L，將結(jié)束值定義為Window.unboundedFollowing：

val winSpec = Window.partitionBy("depName").orderBy("salary") 
             .rangeBetween(300L, Window.unboundedFollowing) 
 
val range_unbounded_df = empsalary.withColumn("max_salary", max("salary").over(winSpec)) 
 
range_unbounded_df.show() 

輸出：

+---------+-----+------+----------+ 
|  depName|empNo|salary|max_salary| 
+---------+-----+------+----------+ 
|  develop|    7|  4200|      6000| 
|  develop|    9|  4500|      6000| 
|  develop|   10|  5200|      6000| 
|  develop|   11|  5200|      6000| 
|  develop|    8|  6000|      null| 
|    sales|    3|  4800|      null| 
|    sales|    4|  4800|      null| 
|    sales|    1|  5000|      null| 
|personnel|    5|  3500|      3900| 
|personnel|    2|  3900|      null| 
+---------+-----+------+----------+ 

因此，對于depname =人員，薪水=3500。窗口將是(開始：3500 + 300 = 3800，結(jié)束：無邊界)。 因此，此范圍內(nèi)的最大值為3900(檢查上面的輸出)。

同樣，對于depname = sales，salary = 4800，窗口將為(開始：4800 + 300、5100，結(jié)束：無邊界)。 由于銷售部門的值不大于5100，因此結(jié)果為空。

2.rowsBetween

通過rangeBetween，我們使用排序列的值定義了窗口的開始和結(jié)束。 但是，我們也可以使用相對行位置定義窗口的開始和結(jié)束。

例如，我們要創(chuàng)建一個窗口，其中窗口的開始是當(dāng)前行之前的一行，結(jié)束是當(dāng)前行之后的一行。

定義自定義窗口規(guī)范

val winSpec = Window.partitionBy("depName") 
            .orderBy("salary").rowsBetween(-1, 1) 

應(yīng)用自定義窗口規(guī)范

val rows_between_df = empsalary.withColumn("max_salary", max("salary").over(winSpec)) 
 
rows_between_df.show() 

輸出：

+---------+-----+------+----------+ 
|  depName|empNo|salary|max_salary| 
+---------+-----+------+----------+ 
|  develop|    7|  4200|      4500| 
|  develop|    9|  4500|      5200| 
|  develop|   10|  5200|      5200| 
|  develop|   11|  5200|      6000| 
|  develop|    8|  6000|      6000| 
|    sales|    3|  4800|      4800| 
|    sales|    4|  4800|      5000| 
|    sales|    1|  5000|      5000| 
|personnel|    5|  3500|      3900| 
|personnel|    2|  3900|      3900| 
+---------+-----+------+----------+ 

現(xiàn)在，讓我們嘗試了解輸出。

• 對于depname =開發(fā)，salary = 4500，將定義一個窗口，該窗口在當(dāng)前行之前和之后一行(以綠色突出顯示)。 因此窗口內(nèi)的薪水為(4200、4500、5200)，最高為5200(上面的檢查輸出)。

 

• 同樣，對于depname = sales，salary = 5000，將在當(dāng)前行的前后定義一個窗口。 由于此行之后沒有行，因此該窗口將只有2行(以綠色突出顯示)，其薪水分別為(4800，5000)和max為5000(上面的檢查輸出)。

 

我們還可以像以前使用rangeBetween一樣使用特殊邊界Window.unboundedPreceding，Window.unboundedFollowing和Window.currentRow。

注意：rowsBetween不需要排序，但是我使用它來使每次運(yùn)行的結(jié)果保持一致。