如何使用ScalaAggregateFunction函数进行聚合操作？

函数名称：Scala Aggregate Function。

函数功能：统计单笔销售金额超过500的记录销售金额累计相加求和。

函数示例：

1）定义Scala Aggregate Function类继承至UserDefinedAggregateFunction：

●重载实现方法inputSchema：返回StructType字段（销售金额，浮点类型），作为Sca-laAggregateFunction函数的输入参数；

●重载实现方法BufferSchema：返回StructType字段（大于500的销售额求和值，浮点类型），作为Scala Aggregate Function函数的中间结果的值；

●重载实现方法update：如果读入每行input的第0个元素的值（即销售金额）不为空，而且input的销售金额值大于500，则更新输入行的聚集缓冲区（buffer），buffer更新第0个元素值（即大于500的销售额求和sum值），其累加当前大于500的销售金额值；

●重载实现方法merge：merge用于合并两个聚集缓冲区，将第一个缓冲区大于500的销售金额求和值加上第二个缓冲区大于500的销售金额求和值，并将结果存储到Mutab-leAggregationBuffer；

●重载实现方法initialize：初始化大于500的销售额求和值为0，用于初始化聚集缓冲区（MutableAggregationBuffer）的值；

●重载实现方法deterministic：设置true，在给定输入值的前提下，ScalaAggregateFunc-tion生成一组相同的结果；(www.xing528.com)

●重载实现方法evaluate：Scala Aggregate Function函数的最终计算结果为buffer的第0个元素（即大于500的销售额求和sum值）。这个值基于每一行的聚合缓冲区的值；

●重载实现方法dataType：dataType表示Scala Aggregate Function函数返回值的类型是浮点类型。

2）定义顾客Customer的case class类，其成员变量分别为ID、姓名、销售额、折扣销售额、所在州等信息。

3）构建SparkContext以及SQLContext，导入spark的sqlContext隐式转换类import sql-Context.implicits._，用于将一个RDD隐式转换为一个DataFrame。

4）构建Customer类型的Seq集合变量custs，通过sc的parallelize方法读入custs数据，调用toDF（）方法转换成DataFrame，使用customerDF.printSchema（）打印出customerDF的Schema结构，将customerDF注册成临时表customerTable。

5）创建Scala Aggregate Function实例mysum，在sqlContext.udf中注册mysum的自定义UDAF函数，将大于500的销售金额汇总累加。

6）在临时表customerTable执行查询操作，根据州分组，查询所属州、调用ScalaAg-gregateFunction函数实例mysum，传入销售金额值计算大于500的销售金额求和的累加值，然后使用sqlResult.printSchema（）打印结果的Schema结构，使用sqlResult.show（）打印查询结果。

在本地运行，结果如下所示。