參考

場景

將本地檔案toNGroup.txt中的內容：

hadoop@master:~/resource$ cat toNGroup.txt 
hadoop 29
hadoop 87
hadoop 39
hadoop 27
hadoop 88
spark 29
spark 90
spark 27
spark 84
spark 92
hadoop@master:~/resource$ 

按照第一個欄位分組，然後按照第二個欄位降序排序，取前4位。即正常結果如下顯示：

spark   92
spark   90
spark   84
spark   29
hadoop  88
hadoop  87
hadoop  39
hadoop  29
 

分析

• 將本地資料匯入到hive表中。spark SQL 通過HiveContext可以直接操作 hive倉庫表中的資料
• 通過視窗函式生成一個數字序列，取該序列的前4條資料即可
  spark sql中提供了很多內建的函式，這個與mysql中內建的函式型別相似，大致分為：
  Aggregate functions、Collection functions、Date time functions、Math functions、String functions、UDF functions以及Window functions - 具體內容可以參看[參考]中的相關連結。通過這些函式，資料分析人員可以很方便的對資料進行各種豐富的挖掘。本文主要以row_number函式實現分組排序為例子，體驗視窗函式的使用。row_number函式說明如下：

def row_number(): Column

Window function: returns a sequential number starting at 1 within a window partition. 

嘿，分組、排序在各大電商網站的應用是有多常見啊！

實驗

package main.scala

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.hive.HiveContext

/**
 * 視窗函式實戰
 * 
 * 1、Spark可以通過HiveContext直接操作Hive中的資料，基於HiveContext我們可以使用sql/hql兩種方式編寫SQL語句對Hive
 * 進行操作：建立、刪除表，往表中匯入資料，以及CRUD
 * 2、通過saveAsTable方式把DataFrame中的資料儲存到Hive資料倉庫中
 * 3、通過 HiveContext.table方式直接載入Hive中的表而生成DataFrame
*/
 

object SparkSQLWindowFunctionOps {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("SparkSQLWindowFunctionOps")
    val sc = new SparkContext(conf)

    val hiveContext = new HiveContext(sc)
    hiveContext.sql("use hive")
    hiveContext.sql("DROP TABLE IF EXISTS scores")
    hiveContext.sql("CREATE TABLE  IF NOT EXISTS scores(name STRING,score INT) ROW FORMAT DELIMITED FIELDS TERMINATED BY ' ' LINES TERMINATED BY '\\n'")

    //把要處理的資料匯入到Hive表中
    hiveContext.sql("LOAD DATA LOCAL INPATH '/home/hadoop/resource/toNGroup.txt' INTO TABLE scores" )

    /*
     * 使用自查詢完成目標資料的提取，在目標資料內使用視窗函式row_number來進行分組排序：
     * PARTITION BY ：指定視窗函式分組的Key
     * ORDER BY :分組後進行排序
     */
    val result =  hiveContext.sql("SELECT name,score "
        + "FROM ("
            + "SELECT "
               + "name,"
               + "score, "
               + "row_number() OVER (PARTITION BY name ORDER BY score DESC ) rank"
               + " FROM scores"
        + ")  sub_scores "
        + " WHERE rank <= 4")

     result.show(); //在Driver的控制檯上打印出結果

    hiveContext.sql("DROP TABLE IF EXISTS sortedResultScores")
    result.saveAsTable("sortedResultScores")
  }
}

執行結果

16/06/01 23:22:19 INFO DAGScheduler: ResultStage 3 (show at SparkSQLWindowFunctionOps.scala:46) finished in 6.969 s
16/06/01 23:22:19 INFO DAGScheduler: Job 1 finished: show at SparkSQLWindowFunctionOps.scala:46, took 7.284524 s
+------+-----+
|  name|score|
+------+-----+
| spark|   92|
| spark|   90|
| spark|   84|
| spark|   29|
|hadoop|   88|
|hadoop|   87|
|hadoop|   39|
|hadoop|   29|
+------+-----+

16/06/01 23:22:19 INFO ParseDriver: Parsing command: DROP TABLE IF EXISTS sortedResultScores
16/06/01 23:22:19 INFO ParseDriver: Parse Completed

總結

1、為什麼稱作 Window function呢？

"SELECT name,score "
        + "FROM ("
            + "SELECT "
               + "name,"
               + "score, "
               + "row_number() OVER (PARTITION BY name ORDER BY score DESC ) rank"
               + " FROM scores"
        + ")  sub_scores "
        + " WHERE rank <= 4"

row_number函式作用於一個分割槽（本例中就是 spark與hadoop形成的兩個分割槽），併為該分割槽中的每條記錄生成一個序列號，這樣在外層迴圈就可以通過過濾該序列號（eg、rank<4）而取特定的資料。從功能上來看，row_number為外層查詢操作裡面的記錄，打開了一扇窗戶（寫不下去了，這個說法實在有點勉強 ～～～），暫時就這麼理解吧！

2、spark sql 通過hiveContext直接操作 hive倉庫中的資料 - 這點實在太棒了啊啊啊！