UDF：User Defined Function。顾名思义，用户自定义函数。我们在之前看过spark内置函数，我们也可以自己定义函数，然后在sql语句里来使用。
UDAF：User Defined Aggregate Function。用户自定义聚合函数。
UDF案例：截取字符串长度的函数
代码实现如下：
package sql
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}
object UDF {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
.setMaster("local")
.setAppName("UDF")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
// 构造模拟数据
val names = Array("Leo", "Marry", "Jack", "Tom")
val namesRDD = sc.parallelize(names, 5)
val namesRowRDD = namesRDD.map{name=>Row(name)}
val structType = StructType(Array(StructField("name",StringType,true)))
val namesDF = sqlContext.createDataFrame(namesRowRDD,structType)
// 注册一张names表
namesDF.registerTempTable("names")
sqlContext.udf.register("strLen",(str:String)=>str.length())
// 使用自定义函数
sqlContext.sql("select name,strLen(name) from names")
.collect()
.foreach(println)
}
}
以上代码第22行中开始定义和注册自定义函数：
定义函数：自己写匿名函数
注册函数：SQLContext.udf.register()
运行代码，输出结果如下：
[Leo,3]
[Marry,5]
[Jack,4]
[Tom,3]
UDAF案例：统计每个字符串出现的次数
上面讲解了UDF，其实更多的是针对单行输入，返回一个输出，这里的UDAF，顾名思义，用户自定义聚合函数，UDAF则可以针对多行输入，进行聚合计算，返回一个输出，功能更加强大，下面我们来看看如何开发UDAF自定义函数
首先我们编码统计一个字符串出现的次数，编写一个类StringCount，虽然这个功能很简单，但是我们用它讲解一下如何实现UserDefinedAggregateFunction,StringCount类的实现如下：
package cn.spark.study.sql
import org.apache.spark.sql.expressions.UserDefinedAggregateFunction
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.DataType
import org.apache.spark.sql.expressions.MutableAggregationBuffer
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.types.IntegerType
/**
* @author Administrator
*/
class StringCount extends UserDefinedAggregateFunction {
// inputSchema，指的是，输入数据的类型
def inputSchema: StructType = {
StructType(Array(StructField("str", StringType, true)))
}
// bufferSchema，指的是，中间进行聚合时，所处理的数据的类型
def bufferSchema: StructType = {
StructType(Array(StructField("count", IntegerType, true)))
}
// dataType，指的是，函数返回值的类型
def dataType: DataType = {
IntegerType
}
def deterministic: Boolean = {
true
}
// 为每个分组的数据执行初始化操作
def initialize(buffer: MutableAggregationBuffer): Unit = {
buffer(0) = 0
}
// 指的是，每个分组，有新的值进来的时候，如何进行分组对应的聚合值的计算
def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
buffer(0) = buffer.getAs[Int](0) + 1
}
// 由于Spark是分布式的，所以一个分组的数据，可能会在不同的节点上进行局部聚合，就是update
// 但是，最后一个分组，在各个节点上的聚合值，要进行merge，也就是合并
def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
buffer1(0) = buffer1.getAs[Int](0) + buffer2.getAs[Int](0)
}
// 指的是，一个分组的聚合值，如何通过中间的缓存聚合值，最后返回一个最终的聚合值
def evaluate(buffer: Row): Any = {
buffer.getAs[Int](0)
}
}
我们发现，继承了UserDefinedAggregateFunction这个类后，就会有很多要实现的方法，以上代码方法已经实现，且每个方法的注释在代码中已经有所标注。
每个重写方法的大概作用如下：
initialize方法：我们的分组数据，每一条数据都会执行初始化操作，我们初始化的时候，把对应的count设置为0就可以了。
update方法：后面每一个分组发现一条数据，就会调用一次update，把数据传过来，input即为传来的值，buffer为之前缓存的值。然后每传来一条，就将之前的缓存值累加一次。
merge方法：一个分组可能会分布在不同节点上，最后要对不同节点上分组的聚合值进行merge合并
evaluate方法：最后我们之前merge合并的都是之前的缓存，我们最后要通过缓存值返回一个最终的聚合值
至此，我们的StringCount类编写完成，接下来，我们开始编写UDAF来应用我们的类：
UDAF.scala:
package sql
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}
object UDAF {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
.setMaster("local")
.setAppName("UDAF")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
// 构造模拟数据
val names = Array("Leo", "Marry", "Jack", "Tom", "Tom", "Tom", "Leo")
val namesRDD = sc.parallelize(names, 5)
val namesRowRDD = namesRDD.map{name=>Row(name)}
val structType = StructType(Array(StructField("name",StringType,true)))
val namesDF = sqlContext.createDataFrame(namesRowRDD,structType)
// 注册一张names表
namesDF.registerTempTable("names")
sqlContext.udf.register("strCount",new StringCount)
// 使用自定义函数
sqlContext.sql("select name,strCount(name) from names group by name")
.collect()
.foreach(println)
}
}
运行代码，输出结果如下：
[Jack,1]
[Tom,3]
[Leo,2]
[Marry,1]
从结果可以看出，已经成功统计出每个字符串的次数，至此，我们的UDAF开发完成。
