14.Spark SQL:UDAF自定义聚合函数实战
2017-11-16 21:27
387 查看
UDAF自定义函数实战
UDAF:User Defined Aggregate Function。用户自定义聚合函数。是Spark 1.5.x引入的最新特性。
UDF,其实更多的是针对单行输入,返回一个输出
这里的UDAF,则可以针对多行输入,进行聚合计算,返回一个输出,功能更加强大
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)
}
}
package cn.spark.study.sql
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.StringType
/**
* UDAF:User Defined Aggregate Function。用户自定义聚合函数。是Spark 1.5.x引入的最新特性。
* UDF,其实更多的是针对单行输入,返回一个输出
* 这里的UDAF,则可以针对多行输入,进行聚合计算,返回一个输出,功能更加强大
*
* @author Administrator
*/
object UDAF {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
.setMaster("local")
.set
4000
AppName("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()
sqlContext.udf.register("strCount", new StringCount)
// 使用自定义函数
sqlContext.sql("select name,strCount(name) from names group by name")
.collect()
.foreach(println)
}
}
本地运行结果:
UDAF:User Defined Aggregate Function。用户自定义聚合函数。是Spark 1.5.x引入的最新特性。
UDF,其实更多的是针对单行输入,返回一个输出
这里的UDAF,则可以针对多行输入,进行聚合计算,返回一个输出,功能更加强大
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)
}
}
package cn.spark.study.sql
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.StringType
/**
* UDAF:User Defined Aggregate Function。用户自定义聚合函数。是Spark 1.5.x引入的最新特性。
* UDF,其实更多的是针对单行输入,返回一个输出
* 这里的UDAF,则可以针对多行输入,进行聚合计算,返回一个输出,功能更加强大
*
* @author Administrator
*/
object UDAF {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
.setMaster("local")
.set
4000
AppName("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()
sqlContext.udf.register("strCount", new StringCount)
// 使用自定义函数
sqlContext.sql("select name,strCount(name) from names group by name")
.collect()
.foreach(println)
}
}
本地运行结果:
相关文章推荐
- 14.Spark SQL:UDAF自定义聚合函数实战
- 14.Spark SQL:UDAF自定义聚合函数实战
- 14.Spark SQL:UDAF自定义聚合函数实战
- 14.Spark SQL:UDAF自定义聚合函数实战
- 14.Spark SQL:UDAF自定义聚合函数实战
- 14.Spark SQL:UDAF自定义聚合函数实战
- 14.Spark SQL:UDAF自定义聚合函数实战
- 14.Spark SQL:UDAF自定义聚合函数实战
- 14.Spark SQL:UDAF自定义聚合函数实战
- 14.Spark SQL:UDAF自定义聚合函数实战
- 14.Spark SQL:UDAF自定义聚合函数实战
- 14.Spark SQL:UDAF自定义聚合函数实战
- 14.Spark SQL:UDAF自定义聚合函数实战
- 14.Spark SQL:UDAF自定义聚合函数实战
- 14.Spark SQL:UDAF自定义聚合函数实战
- 14.Spark SQL:UDAF自定义聚合函数实战
- 14.Spark SQL:UDAF自定义聚合函数实战
- Spark SQL 用户自定义函数UDF、用户自定义聚合函数UDAF 教程(Java踩坑教学版)
- Spark SQL 用户自定义函数UDF、用户自定义聚合函数UDAF 教程(Java踩坑教学版)
- 【Spark篇】---SparkSQL中自定义UDF和UDAF,开窗函数的应用