Spark Sql的UDF和UDAF函数


Spark Sql提供了丰富的内置函数供猿友们使用,辣为何还要用户自定义函数呢?实际的业务场景可能很复杂,内置函数hold不住,所以spark sql提供了可扩展的内置函数接口:哥们,你的业务太变态了,我满足不了你,自己按照我的规范去定义一个sql函数,该怎么折腾就怎么折腾!

这里还是先以Scala实现一个简单的hello world级别的小样为例,来体验udf与udaf的使用好了。

问题

将如下数组:

val bigData = Array("Spark","Hadoop","Flink","Spark","Hadoop","Flink",
"Spark","Hadoop","Flink","Spark","Hadoop","Flink")

中的字符分组聚合并计算出每个字符的长度及字符出现的个数。正常结果 
如下:

+------+-----+------+
|  name|count|length|
+------+-----+------+
| Spark|    4|     5|
| Flink|    4|     5|
|Hadoop|    4|     6|
+------+-----+------+

注:‘spark’ 这个字符的长度为5 ,共出现了4次。

分析

  • 自定义个一个求字符串长度的函数 
    自定义的sql函数,与scala中的普通函数一样,只不过在使用上前者需要先在sqlContext中进行注册。
  • 自定义一个聚合函数 
    按照字符串名称分组后,调用自定义的聚合函数实现累加。 
    啊,好抽象,直接看代码吧!

代码

package com.hand.datasafe

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Row, SQLContext, SparkSession}
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.StringType

/**
  * Spark SQL UDAF:user defined aggregation function
  * UDF: 函数的输入是一条具体的数据记录,实现上讲就是普通的scala函数-只不过需要注册
  * UDAF:用户自定义的聚合函数,函数本身作用于数据集合,能够在具体操作的基础上进行自定义操作
  */
object SparkSQLUDF {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder().appName("datasafe").master("local").getOrCreate()

    val bigData = Array("Spark", "Hadoop", "Flink", "Spark", "Hadoop", "Flink", "Spark", "Hadoop", "Flink", "Spark", "Hadoop", "Flink")
    val bigDataRDD = spark.sparkContext.parallelize(bigData)

    val bigDataRowRDD: RDD[Row] = bigDataRDD.map(line => Row(line))
    val structType = StructType(Array(StructField("name", StringType, true)))
    val bigDataDF = spark.createDataFrame(bigDataRowRDD, structType)
    bigDataDF.printSchema()
    bigDataDF.createTempView("bigDataTable")

    /*
     * 通过saprk注册UDF,在scala2.1.x版本UDF函数最多可以接受22个输入参数
     */
    spark.udf.register("computeLength", (input: String) => input.length)
    spark.sql("select name,computeLength(name)  as length from bigDataTable").show

    //while(true){}

    spark.udf.register("wordCount", new MyUDAF)
    spark.sql("select name,wordCount(name) as count,computeLength(name) as length from bigDataTable group by name ").show
    spark.sql("select name,wordCount(name) as count,computeLength(name) as length from bigDataTable group by name ").printSchema()

  }
}
package com.hand.datasafe

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

/**
  * 用户自定义函数
  */
class MyUDAF extends UserDefinedAggregateFunction
{
  /**
    * 指定具体的输入数据的类型
    * 自段名称随意:Users can choose names to identify the input arguments - 这里可以是“name”,或者其他任意串
    */
  override def inputSchema:StructType = StructType(Array(StructField("name",StringType,true)))

  /**
    * 在进行聚合操作的时候所要处理的数据的中间结果类型
    */
  override def bufferSchema:StructType = StructType(Array(StructField("count",IntegerType,true)))

  /**
    * 返回类型
    */
  override def dataType:DataType = IntegerType

  /**
    * whether given the same input,
    * always return the same output
    * true: yes
    */
  override def deterministic:Boolean = true

  /**
    * Initializes the given aggregation buffer
    */
  override def initialize(buffer:MutableAggregationBuffer):Unit = {buffer(0)=0}

  /**
    * 在进行聚合的时候,每当有新的值进来,对分组后的聚合如何进行计算
    * 本地的聚合操作,相当于Hadoop MapReduce模型中的Combiner
    */
  override def update(buffer:MutableAggregationBuffer,input:Row):Unit={
    buffer(0) = buffer.getInt(0)+1
  }

  /**
    * 最后在分布式节点进行local reduce完成后需要进行全局级别的merge操作
    */
  override def merge(buffer1:MutableAggregationBuffer,buffer2:Row):Unit={
    buffer1(0) = buffer1.getInt(0)+buffer2.getInt(0)
  }

  /**
    * 返回UDAF最后的计算结果
    */
  override def evaluate(buffer:Row):Any = buffer.getInt(0)
}

总结

    • 呼叫spark大神升级udaf实现 
      为了自己实现一个sql聚合函数,我需要继承UserDefinedAggregateFunction并实现8个抽象方法!8个方法啊!what’s a disaster ! 然而,要想在sql中完成符合特定业务场景的聚合类(a = aggregation)功能,就得udaf。 
      怎么理解MutableAggregationBuffer呢?就是存储中间结果的,聚合就意味着多条记录的累加等操作。


注意!

本站转载的文章为个人学习借鉴使用,本站对版权不负任何法律责任。如果侵犯了您的隐私权益,请联系我们删除。



 
© 2014-2019 ITdaan.com 粤ICP备14056181号