sparkSQL--over窗口函數（實戰案例）

一、over（窗口函數）

指的是對多行數據進行處理返回普通列和聚合列的過程

詳細語法：

窗口函數sql語法：窗口函數名（）over (partition by 划分窗口字段 order by 窗口內的排序規則 rows between (start,end))

窗口函數分類：

• 聚合窗口函數 aggregate 聚合類
• 排名窗口函數 ranking 排名類
• 數據分析窗口函數 analytic 分析類

參考鏈接：https://www.cnblogs.com/abc8023/p/10910741.html

Function Type	SQL	DataFrame API	Description
Ranking	rank	rank	rank值可能是不連續的
Ranking	dense_rank	denseRank	rank值一定是連續的
Ranking	percent_rank	percentRank	相同的分組中 (rank -1) / ( count(score) - 1 )
Ranking	ntile	ntile	將同一組數據循環的往n個桶中放，返回對應的桶的index，index從1開始
Ranking	row_number	rowNumber	很單純的行號，類似excel的行號
Analytic	cume_dist	cumeDist
Analytic	first_value	firstValue	相同的分組中最小值
Analytic	last_value	lastValue	相同的分組中最大值
Analytic	lag	lag	取前n行數據
Analytic	lead	lead	取后n行數據
Aggregate	min	min	最小值
Aggregate	max	max	最大值
Aggregate	sum	sum	求和
Aggregate	avg	avg	求平均

二、具體用法如下

count(...) over(partition by ... order by ...) --求分組后的總數。

sum(...) over(partition by ... order by ...) --求分組后的和。

max(...) over(partition by ... order by ...) --求分組后的最大值。

min(...) over(partition by ... order by ...) --求分組后的最小值。

avg(...) over(partition by ... order by ...) --求分組后的平均值。

rank() over(partition by ... order by ...) --rank值可能是不連續的。

dense_rank() over(partition by ... order by ...) --rank值是連續的。

first_value(...) over(partition by ... order by ...) --求分組內的第一個值。

last_value(...) over(partition by ... order by ...) --求分組內的最后一個值。

lag() over(partition by ... order by ...) --取出前n行數據。

lead() over(partition by ... order by ...) --取出后n行數據。

ratio_to_report() over(partition by ... order by ...) --Ratio_to_report() 括號中就是分子，over() 括號中就是分母。

percent_rank() over(partition by ... order by ...)

三、應用案例

問題

某app訪問頁面的日志詳細記錄字段如下：day, user_id, page_id, time

求某天每個用戶訪問頁面次數前10的頁面。

("2018-01-01",1,"www.baidu.com","10:01"),
("2018-01-01",2,"www.baidu.com","10:01"),
("2018-01-01",1,"www.sina.com","10:01"),
("2018-01-01",3,"www.baidu.com","10:01"),
("2018-01-01",3,"www.baidu.com","10:01"),
("2018-01-01",1,"www.sina.com","10:01")

思路

1. 每個用戶訪問不同頁面的次數

select user_id,page_id,count(page_id) from t_log group by user_id, page_id

+-------+-------------+-----+---+---|
|user_id| page_id|count|
+-------+-------------+-----+----+--|
| 2|www.baidu.com| 1|
| 3|www.baidu.com| 2|
| 1|www.baidu.com| 1|
| 1| www.sina.com| 2|
+-------+-------------+-----+----+--|

2. 對每個用戶點擊頁面次數降序排列，並且使用窗口函數中的排名函數，對點擊頁面進行排名

w1:

| 1| www.sina.com| 2| 1

| 1|www.baidu.com| 1| 2

w2:

| 2|www.baidu.com| 1| 1

w3:

| 3|www.baidu.com| 2| 1

+-------+-------------+-----+----+
|user_id| page_id|count|rank|
+-------+-------------+-----+----+
| 1| www.sina.com| 2| 1|
| 1|www.baidu.com| 1| 2|
| 3|www.baidu.com| 2| 1|
| 2|www.baidu.com| 1| 1|
+-------+-------------+-----+----+

3. 獲得每個用戶訪問次數前10的頁面

where rank <= 10

+-------+-------------+-----+----+
|user_id| page_id|count|rank|
+-------+-------------+-----+----+
| 1| www.sina.com| 2| 1|
| 1|www.baidu.com| 1| 2|
| 3|www.baidu.com| 2| 1|
| 2|www.baidu.com| 1| 1|
+-------+-------------+-----+----+

代碼

package method

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.expressions.Window

object SQLDemo3 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName("sql operation").master("local[*]").getOrCreate()
    val rdd = spark.sparkContext.makeRDD(
      List(
        ("2018-01-01",1,"www.baidu.com","10:01"),
        ("2018-01-01",2,"www.baidu.com","10:01"),
        ("2018-01-01",1,"www.sina.com","10:01"),
        ("2018-01-01",3,"www.baidu.com","10:01"),
        ("2018-01-01",3,"www.baidu.com","10:01"),
        ("2018-01-01",1,"www.sina.com","10:01")
      )
    )
    import spark.implicits._
    val df = rdd.toDF("day","user_id","page_id","time")
  df.createTempView("t_log")
    //注意：""" 包裹內容 “”“自動進行字符串的拼接
    spark
      .sql(
        """
          |select *
          |from
          | (select user_id,page_id, num,
          |   rank() over(partition by user_id order by num desc) as rank
          |   from
          |     (select
          |       user_id,
          |       page_id,
          |       count(page_id) as num
          |       from t_log
          |       group by user_id,page_id))
          | where rank <= 10
          |
          |""".stripMargin
      )
      .show()
  spark.stop()
  }
}
//結果
+-------+-------------+---+----+
|user_id|      page_id|num|rank|
+-------+-------------+---+----+
|      1| www.sina.com|  2|   1|
|      1|www.baidu.com|  1|   2|
|      3|www.baidu.com|  2|   1|
|      2|www.baidu.com|  1|   1|
+-------+-------------+---+----+