如何在spark里面使用窗口函数

在大数据分析中,窗口函数最常见的应用场景就是对数据进行分组后,求组内数据topN的需求,如果没有窗口函数,实现这样一个需求还是比较复杂的,不过现在大多数标准SQL中都支持这样的功能,今天我们就来学习下如何在spark sql使用窗口函数来完成一个分组求TopN的需求。

思路分析:

在spark sql中有两种方式可以实现:

(1)使用纯spark sql的方式。

(2)spark的编程api来实现。

虽然有两种形式,但底层原理都一样,借助了spark里面的window算子,我们先来看下纯sql的实现方式,其代码如下:

   def main(args: Array[String]): Unit = {
    val spark = SparkSession      .builder().master("local[1]")      .appName("Spark SQL basic example")      .getOrCreate()
    import spark.implicits._    val df = spark.sparkContext.parallelize(Seq(      (1,"2019-03-01","河北","ios"),      (1,"2019-06-01","北京","Android"),      (1,"2019-07-01","北京","Android"),      (1,"2019-07-01","北京","Android"),      (2,"2019-04-01","河南","Android"),      (2,"2019-06-01","山西","ios"),      (2,"2019-08-01","湖南","ios")    )).toDF("id", "date", "address","device")//转化df的三列数据s
    df.createOrReplaceTempView("login")
    //先对组内数据,进行排序    val s2=spark.sql("select id, date,address,device, rank() over (partition by id order by date desc ) as rank from login")    s2.createOrReplaceTempView("login2")    //取top N    val s3=spark.sql("select * from login2 where rank=1")
    s3.show()
  }

我们来看下输出结果如下:

+---+----------+-------+-------+----+| id|      date|address| device|rank|+---+----------+-------+-------+----+|  1|2019-07-01|     北京|Android|   1||  1|2019-07-01|     北京|Android|   1||  2|2019-08-01|     湖南|    ios|   1|+---+----------+-------+-------+----+

注意这里,我为了保持整洁,没有使用嵌套的子查询,而是在s3处,又过滤了一下结果。 我们看到,在sql中我们借助使用了rank函数,因为id=1的,最新日期有两个一样的,所以rank相等, 故最终结果返回了三条数据,到这里有的朋友可能就有疑问了,我只想对每组数据取topN,比如每组只取一条应该怎么控制,现在某组可能会返回2条,虽然意义上没错,但总觉得不太好,那么能不能实现呢?

答案是可以的,这就涉及到关于排名函数的介绍,我们这里只介绍常用的三种,分别是:

(1)rank

(2)row_number

(3)dense_rank

这次,我们用代码实现上面的需求,并观察上面上个函数生成rank值的区别,代码如下:

  def main(args: Array[String]): Unit = {
    val spark = SparkSession      .builder().master("local[1]")      .appName("Spark SQL basic example")      .getOrCreate()
    import spark.implicits._    val df = spark.sparkContext.parallelize(Seq(      (1,"2019-03-01","河北","ios"),      (1,"2019-06-01","北京","Android"),      (1,"2019-07-01","北京","Android"),      (1,"2019-07-01","北京","Android"),      (2,"2019-04-01","河南","Android"),      (2,"2019-06-01","山西","ios"),      (2,"2019-08-01","湖南","ios")    )).toDF("id", "date", "address","device")//转化df的三列数据s
//    df.createOrReplaceTempView("login")
    val s2=Window.partitionBy("id").orderBy(col("date").desc)    df.withColumn("rank",rank().over(s2))//生成rank值可以重复但不一定连续      .withColumn("dense_rank",dense_rank().over(s2))//生成rank值可以重复但是连续      .withColumn("row_number",row_number().over(s2))//生成的rank值不重复但是连续      .show()

  }

ok,我们看下输出结果:

+---+----------+-------+-------+----+----------+----------+| id|      date|address| device|rank|dense_rank|row_number|+---+----------+-------+-------+----+----------+----------+|  1|2019-07-01|     北京|Android|   1|         1|         1||  1|2019-07-01|     北京|Android|   1|         1|         2||  1|2019-06-01|     北京|Android|   3|         2|         3||  1|2019-03-01|     河北|    ios|   4|         3|         4||  2|2019-08-01|     湖南|    ios|   1|         1|         1||  2|2019-06-01|     山西|    ios|   2|         2|         2||  2|2019-04-01|     河南|Android|   3|         3|         3|+---+----------+-------+-------+----+----------+----------+

注意看输出数据的前三行,观察后面的值,我们能够发现上面三个函数的区别是:

(1)rank (生成rank值可以重复但不一定连续)

(2)row_number (生成rank值可以重复但是连续)

(3)dense_rank (生成的rank值不重复但是连续)

了解上面的区别后,我们再回到刚才的那个问题,如何取Top1的时候,每组只返回一条数据?

答案就是使用row_number进行过滤,如下,对上面的代码稍加改造即可:

    val s2=Window.partitionBy("id").orderBy(col("date").desc)    df.withColumn("rank",rank().over(s2))//生成rank值可以重复但不一定连续      .withColumn("dense_rank",dense_rank().over(s2))//生成rank值可以重复但是连续      .withColumn("row_number",row_number().over(s2))//生成的rank值不重复但是连续      .where("row_number=1")//新增代码      .show()

结果如下:

+---+----------+-------+-------+----+----------+----------+| id|      date|address| device|rank|dense_rank|row_number|+---+----------+-------+-------+----+----------+----------+|  1|2019-07-01|     北京|Android|   1|         1|         1||  2|2019-08-01|     湖南|    ios|   1|         1|         1|+---+----------+-------+-------+----+----------+----------+

在sql里面也一样,只要把rank函数换成row_number函数即可,这里就不在演示了,感兴趣的同学可以自己尝试下。

在spark的窗口函数里面,上面的应用场景属于比较常见的case,当然spark窗口函数的功能要比上面介绍的要丰富的多,这里就不在介绍了,想学习的同学可以参考下面的这个链接:

https://databricks.com/blog/2015/07/15/introducing-window-functions-in-spark-sql.html

原文发布于微信公众号 - 我是攻城师(woshigcs)

原文发表时间:2019-06-21

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

扫码关注云+社区

领取腾讯云代金券