SparkSQL开窗函数 row_number()

开始编写我们的统计逻辑,使用row_number()函数
先说明一下,row_number()开窗函数的作用
其实就是给每个分组的数据,按照其排序顺序,打上一个分组内行号
比如说,有一个分组20151001,里面有三条数据,1122,1121,1124
那么对这个分组的每一行使用row_number()开窗函数以后,三行依次会获得一个组内的行号
行号从1开始递增,比如1122,1 1121,2 1124,3

row_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
    */
}

猜你喜欢

转载自www.cnblogs.com/suixingc/p/sparksql-kai-chuang-han-shu---rownumber.html