一、简单聚合

1.1 数据准备

// 需要导入 spark sql 内置的函数包
import org.apache.spark.sql.functions._

val spark = SparkSession.builder().appName("aggregations").master("local[2]").getOrCreate()
val empDF = spark.read.json("/usr/file/json/emp.json")
// 注册为临时视图,用于后面演示 SQL 查询
empDF.createOrReplaceTempView("emp")
empDF.show()

注:emp.json 可以从本仓库的resources 目录下载。

1.2 count

// 计算员工人数
empDF.select(count("ename")).show()

1.3 countDistinct

// 计算姓名不重复的员工人数
empDF.select(countDistinct("deptno")).show()

1.4 approx_count_distinct

通常在使用大型数据集时,你可能关注的只是近似值而不是准确值,这时可以使用 approx_count_distinct 函数,并可以使用第二个参数指定最大允许误差。

empDF.select(approx_count_distinct ("ename",0.1)).show()

1.5 first & last

获取 DataFrame 中指定列的第一个值或者最后一个值。

empDF.select(first("ename"),last("job")).show()

1.6 min & max

获取 DataFrame 中指定列的最小值或者最大值。

empDF.select(min("sal"),max("sal")).show()

1.7 sum & sumDistinct

求和以及求指定列所有不相同的值的和。

empDF.select(sum("sal")).show()
empDF.select(sumDistinct("sal")).show()

1.8 avg

内置的求平均数的函数。

empDF.select(avg("sal")).show()

1.9 数学函数

Spark SQL 中还支持多种数学聚合函数,用于通常的数学计算,以下是一些常用的例子:

// 1.计算总体方差、均方差、总体标准差、样本标准差
empDF.select(var_pop("sal"), var_samp("sal"), stddev_pop("sal"), stddev_samp("sal")).show()

// 2.计算偏度和峰度
empDF.select(skewness("sal"), kurtosis("sal")).show()

// 3. 计算两列的皮尔逊相关系数、样本协方差、总体协方差。(这里只是演示,员工编号和薪资两列实际上并没有什么关联关系)
empDF.select(corr("empno", "sal"), covar_samp("empno", "sal"),covar_pop("empno", "sal")).show()

1.10 聚合数据到集合

scala>  empDF.agg(collect_set("job"), collect_list("ename")).show()

输出:
+--------------------+--------------------+
|    collect_set(job)| collect_list(ename)|
+--------------------+--------------------+
|[MANAGER, SALESMA...|[SMITH, ALLEN, WA...|
+--------------------+--------------------+

二、分组聚合

2.1 简单分组

empDF.groupBy("deptno", "job").count().show()
//等价 SQL
spark.sql("SELECT deptno, job, count(*) FROM emp GROUP BY deptno, job").show()

输出:
+------+---------+-----+
|deptno|      job|count|
+------+---------+-----+
|    10|PRESIDENT|    1|
|    30|    CLERK|    1|
|    10|  MANAGER|    1|
|    30|  MANAGER|    1|
|    20|    CLERK|    2|
|    30| SALESMAN|    4|
|    20|  ANALYST|    2|
|    10|    CLERK|    1|
|    20|  MANAGER|    1|
+------+---------+-----+

2.2 分组聚合

empDF.groupBy("deptno").agg(count("ename").alias("人数"), sum("sal").alias("总工资")).show()
// 等价语法
empDF.groupBy("deptno").agg("ename"->"count","sal"->"sum").show()
// 等价 SQL
spark.sql("SELECT deptno, count(ename) ,sum(sal) FROM emp GROUP BY deptno").show()

输出:
+------+----+------+
|deptno|人数|总工资|
+------+----+------+
|    10|   3|8750.0|
|    30|   6|9400.0|
|    20|   5|9375.0|
+------+----+------+

三、自定义聚合函数

Scala 提供了两种自定义聚合函数的方法,分别如下:

  • 有类型的自定义聚合函数,主要适用于 DataSet;
  • 无类型的自定义聚合函数,主要适用于 DataFrame。

以下分别使用两种方式来自定义一个求平均值的聚合函数,这里以计算员工平均工资为例。两种自定义方式分别如下:

3.1 有类型的自定义函数

import org.apache.spark.sql.expressions.Aggregator
import org.apache.spark.sql.{Encoder, Encoders, SparkSession, functions}

// 1.定义员工类,对于可能存在 null 值的字段需要使用 Option 进行包装
case class Emp(ename: String, comm: scala.Option[Double], deptno: Long, empno: Long,
               hiredate: String, job: String, mgr: scala.Option[Long], sal: Double)

// 2.定义聚合操作的中间输出类型
case class SumAndCount(var sum: Double, var count: Long)

/* 3.自定义聚合函数
 * @IN  聚合操作的输入类型
 * @BUF reduction 操作输出值的类型
 * @OUT 聚合操作的输出类型
 */
object MyAverage extends Aggregator[Emp, SumAndCount, Double] {

    // 4.用于聚合操作的的初始零值
    override def zero: SumAndCount = SumAndCount(0, 0)

    // 5.同一分区中的 reduce 操作
    override def reduce(avg: SumAndCount, emp: Emp): SumAndCount = {
        avg.sum += emp.sal
        avg.count += 1
        avg
    }

    // 6.不同分区中的 merge 操作
    override def merge(avg1: SumAndCount, avg2: SumAndCount): SumAndCount = {
        avg1.sum += avg2.sum
        avg1.count += avg2.count
        avg1
    }

    // 7.定义最终的输出类型
    override def finish(reduction: SumAndCount): Double = reduction.sum / reduction.count

    // 8.中间类型的编码转换
    override def bufferEncoder: Encoder[SumAndCount] = Encoders.product

    // 9.输出类型的编码转换
    override def outputEncoder: Encoder[Double] = Encoders.scalaDouble
}

object SparkSqlApp {

    // 测试方法
    def main(args: Array[String]): Unit = {

        val spark = SparkSession.builder().appName("Spark-SQL").master("local[2]").getOrCreate()
        import spark.implicits._
        val ds = spark.read.json("file/emp.json").as[Emp]

        // 10.使用内置 avg() 函数和自定义函数分别进行计算,验证自定义函数是否正确
        val myAvg = ds.select(MyAverage.toColumn.name("average_sal")).first()
        val avg = ds.select(functions.avg(ds.col("sal"))).first().get(0)

        println("自定义 average 函数 : " + myAvg)
        println("内置的 average 函数 : " + avg)
    }
}

自定义聚合函数需要实现的方法比较多,这里以绘图的方式来演示其执行流程,以及每个方法的作用:

关于 zero,reduce,merge,finish 方法的作用在上图都有说明,这里解释一下中间类型和输出类型的编码转换,这个写法比较固定,基本上就是两种情况:

  • 自定义类型 Case Class 或者元组就使用 Encoders.product 方法;
  • 基本类型就使用其对应名称的方法,如 scalaBytescalaFloatscalaShort 等,示例如下:
override def bufferEncoder: Encoder[SumAndCount] = Encoders.product
override def outputEncoder: Encoder[Double] = Encoders.scalaDouble

3.2 无类型的自定义聚合函数

理解了有类型的自定义聚合函数后,无类型的定义方式也基本相同,代码如下:

import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types._
import org.apache.spark.sql.{Row, SparkSession}

object MyAverage extends UserDefinedAggregateFunction {
  // 1.聚合操作输入参数的类型,字段名称可以自定义
  def inputSchema: StructType = StructType(StructField("MyInputColumn", LongType) :: Nil)

  // 2.聚合操作中间值的类型,字段名称可以自定义
  def bufferSchema: StructType = {
    StructType(StructField("sum", LongType) :: StructField("MyCount", LongType) :: Nil)
  }

  // 3.聚合操作输出参数的类型
  def dataType: DataType = DoubleType

  // 4.此函数是否始终在相同输入上返回相同的输出,通常为 true
  def deterministic: Boolean = true

  // 5.定义零值
  def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer(0) = 0L
    buffer(1) = 0L
  }

  // 6.同一分区中的 reduce 操作
  def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    if (!input.isNullAt(0)) {
      buffer(0) = buffer.getLong(0) + input.getLong(0)
      buffer(1) = buffer.getLong(1) + 1
    }
  }

  // 7.不同分区中的 merge 操作
  def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0)
    buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)
  }

  // 8.计算最终的输出值
  def evaluate(buffer: Row): Double = buffer.getLong(0).toDouble / buffer.getLong(1)
}

object SparkSqlApp {

  // 测试方法
  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder().appName("Spark-SQL").master("local[2]").getOrCreate()
    // 9.注册自定义的聚合函数
    spark.udf.register("myAverage", MyAverage)

    val df = spark.read.json("file/emp.json")
    df.createOrReplaceTempView("emp")

    // 10.使用自定义函数和内置函数分别进行计算
    val myAvg = spark.sql("SELECT myAverage(sal) as avg_sal FROM emp").first()
    val avg = spark.sql("SELECT avg(sal) as avg_sal FROM emp").first()

    println("自定义 average 函数 : " + myAvg)
    println("内置的 average 函数 : " + avg)
  }
}

参考资料

  1. Matei Zaharia, Bill Chambers . Spark: The Definitive Guide[M] . 2018-02

更多大数据系列文章可以参见 GitHub 开源项目大数据入门指南

Spark 系列(十一)—— Spark SQL 聚合函数 Aggregations的更多相关文章

  1. Spark学习之路(十一)—— Spark SQL 聚合函数 Aggregations

    一.简单聚合 1.1 数据准备 // 需要导入spark sql内置的函数包 import org.apache.spark.sql.functions._ val spark = SparkSess ...

  2. SQL 聚合函数

    SQL聚合函数 MAX---最大值 MIN--最小值 AVG--平均值 SUM--求和 COUNT--记录的条数 EXample: --从MyStudent表中查询最大年龄,最小年龄,平均年龄,年龄的 ...

  3. SQL Server数据库--》top关键字,order by排序,distinct去除重复记录,sql聚合函数,模糊查询,通配符,空值处理。。。。

    top关键字:写在select后面 字段的前面 比如你要显示查询的前5条记录,如下所示: select top 5 * from Student 一般情况下,top是和order by连用的 orde ...

  4. sql 聚合函数、排序方法详解

    聚合函数 count,max,min,avg,sum... select count (*) from T_Employee select Max(FSalary) from T_Employee 排 ...

  5. C#写的SQL聚合函数

    SQL Server 字符串连接聚合函数. 注册程序集: 拷贝“SqlStrConcate.dll”至<sql安装根目录>/MSSQL.1/MSSQL/Binn目录下,执行下面的SQL: ...

  6. Sql Server的艺术(三) SQL聚合函数的应用

    SQL提供的聚合函数有求和,最大值,最小值,平均值,计数函数等. 聚合函数及其功能: 函数名称 函数功能 SUM() 返回选取结果集中所有值的总和 MAX() 返回选取结果集中所有值的最大值 MIN( ...

  7. sql 聚合函数用法,及执行顺序

    聚合函数无法用在where子句中 , 聚合函数包括count avg sum min max 子句执行顺序from -> where -> group by -> having -& ...

  8. sql 聚合函数和group by 联合使用

    原文 很多时候单独使用聚合函数的时候觉得很容易,求个平均值,求和,求个数等,但是和分组一起用就有点混淆了,好记性不如烂笔头,所以就记下来以后看看. 常用聚合函数罗列 1 AVG() - 返回平均值 C ...

  9. Spark系列—01 Spark集群的安装

    一.概述 关于Spark是什么.为什么学习Spark等等,在这就不说了,直接看这个:http://spark.apache.org, 我就直接说一下Spark的一些优势: 1.快 与Hadoop的Ma ...

  10. Spark系列—02 Spark程序牛刀小试

    一.执行第一个Spark程序 1.执行程序 我们执行一下Spark自带的一个例子,利用蒙特·卡罗算法求PI: 启动Spark集群后,可以在集群的任何一台机器上执行一下命令: /home/spark/s ...

随机推荐

  1. PHP函数、数组

    PHP函数 PHP的真正威力源自于它的函数函数是通过调用函数来执行的,可以在页面的任何位置调用函数 PHP函数准则:函数名称以字母或下划线开头(不能以数字开头)函数的名称应该提示出它的功能 四要素:返 ...

  2. 《BI那点儿事》浅析十三种常用的数据挖掘的技术

    一.前沿 数据挖掘就是从大量的.不完全的.有噪声的.模糊的.随机的数据中,提取隐含在其中的.人们事先不知道的但又是潜在有用的信息和知识的过程.数据挖掘的任务是从数据集中发现模式,可以发现的模式有很多种 ...

  3. HTTP协议——学习资料小结

    嗯,这几天回头再次的学习Servlet的知识点,觉得HTTP协议的内容是相当重要的,现在虽然知道浏览器与应用程序的交互离不开它,但是怎么将信息从浏览器传输到服务器的这个知识点还是一个盲点.于是从网上找 ...

  4. Android小项目之十二 设置中心的界面

    ------- 源自梦想.永远是你IT事业的好友.只是勇敢地说出我学到! ---------- 按惯例,写在前面的:可能在学习Android的过程中,大家会和我一样,学习过大量的基础知识,很多的知识点 ...

  5. JPA学习---第五节:日期和枚举等字段类型的JPA映射

    1.在上一节可在数据库中看到创建出来的表和字段,是通过 Entity bean 来创建的,而创建表名和字段名的规则是怎样的? 有类,代码如下: package learn.jpa.bean; impo ...

  6. phpStorm设置显示代码行号

    File->Settings

  7. Chapter 1 An Overview of Computers and Programming Languages

    Babylon巴比伦 loom织布机 weaver, WHO uses loom to work census: to count the population tabulate: make into ...

  8. 使用VB6读取数据库资源并发送邮件(原创)

    Private Sub Form_Load() Call conndb End Sub Private Function conndb() Dim cn As New ADODB.Connection ...

  9. enq: TM - contention一例

    今天下午,有台服务器出现异常,响应特别慢,io等待奇高,awr top 5事件如下: 经回查ash,找到了造成这些事件的sql语句,如下: select * from v$active_session ...

  10. Jsonpath的基本使用

    JSONPath - 是xpath在json的应用. xml最大的优点就有大量的工具可以分析,转换,和选择性的提取文档中的数据.XPath是这些最强大的工具之一.   如果可以使用xpath来解析js ...