开始编写我们的统计逻辑,使用row_number()函数
先说明一下,row_number()开窗函数的作用
其实就是给每个分组的数据,按照其排序顺序,打上一个分组内行号
比如说,有一个分组20151001,里面有三条数据,1122,1121,1124
那么对这个分组的每一行使用row_number()开窗函数以后,三行依次会获得一个组内的行号
行号从1开始递增,比如1122,1 1121,2 1124,3row_number()开窗函数的语法说明
首先可以在select查询时,使用row_number()函数
其次,row_number()函数后面先跟上over关键字
然后括号中是partition by也就是根据哪个字段进行分组
其次是可以用order by进行组内排序 然后row_number()就可以给每个组内的行,一个组内行号
RowNumberWindowFunc.scala
package com.UDF.row_numberFUNC
import org.apache.spark.sql.{SaveMode, SparkSession}
object RowNumberWindowFunc extends App {
val spark = SparkSession
.builder()
.appName("RowNumberWindowFunc")
.master("local[2]")
.getOrCreate()
//创建销售额表,sales表
spark.sql("drop table if exists sales")
spark.sql("create table if not exists sales ("
+ "product string, "
+ "category string, "
+ "revenue bigint)")
spark.sql("load data "
+ "load inpath '/usr/local/data'"
+ "into table sales")
//开始编写我们的统计逻辑,使用row_number()函数
//先说明一下,row_number()开窗函数的作用
//其实就是给每个分组的数据,按照其排序顺序,打上一个分组内行号
//比如说,有一个分组20151001,里面有三条数据,1122,1121,1124
//那么对这个分组的每一行使用row_number()开窗函数以后,三行依次会获得一个组内的行号
//行号从1开始递增,比如1122,1 1121,2 1124,3
val top3SalesDF = spark.sql(""
+ "select product,category,revenue"
+ "from ("
+ "select product,category,revenue,"
//row_number()开窗函数的语法说明
//首先可以在select查询时,使用row_number()函数
//其次,row_number()函数后面先跟上over关键字
//然后括号中是partition by也就是根据哪个字段进行分组
//其次是可以用order by进行组内排序
//然后row_number()就可以给每个组内的行,一个组内行号
+ "row_number() over (partition by catefory order by revenue desc ) rank "
+ " from sales) tmp_sales "
+ "where rank <= 3")
//将魅族排名前三的数据,保存到一个表中
spark.sql("drop table if exists top3_sales")
top3SalesDF.write //保存,要用write开头
.mode(SaveMode.Overwrite) //覆盖模式
.format("hive") //格式hive (hive默认格式,数据文件纯文本无压缩存储)
.saveAsTable("top3_sales") //做为表保存
/**
* format支持的格式有:
* hive (hive默认格式,数据文件纯文本无压缩存储)
* parquet (spark默认采用格式)
* orc
* json
* csv
* text (若用saveAsTable只能保存一个列的df)
* jdbc
* libsvm
*/
}