Spark学习之路 (十九)SparkSQL的自定义函数UDF[转]

在Spark中,也支持Hive中的自定义函数。自定义函数大致可以分为三种:

UDF(User-Defined-Function),即最基本的自定义函数,类似to_char,to_date等
UDAF(User- Defined Aggregation Funcation),用户自定义聚合函数,类似在group by之后使用的sum,avg等
UDTF(User-Defined Table-Generating Functions),用户自定义生成函数,有点像stream里面的flatMap
自定义一个UDF函数需要继承UserDefinedAggregateFunction类,并实现其中的8个方法

示例

import org.apache.spark.sql.Row
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types.{DataType, StringType, StructField, StructType}

object GetDistinctCityUDF extends UserDefinedAggregateFunction{
  /**
    * 输入的数据类型
    * */
  override def inputSchema: StructType = StructType(
    StructField("status",StringType,true) :: Nil
  )
  /**
    * 缓存字段类型
    * */
  override def bufferSchema: StructType = {
    StructType(
      Array(
        StructField("buffer_city_info",StringType,true)
      )
    )
  }
/**
  * 输出结果类型
  * */
  override def dataType: DataType = StringType
/**
  * 输入类型和输出类型是否一致
  * */
  override def deterministic: Boolean = true
/**
  * 对辅助字段进行初始化
  * */
  override def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer.update(0,"")
  }
/**
  *修改辅助字段的值
  * */
  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    //获取最后一次的值
    var last_str = buffer.getString(0)
    //获取当前的值
    val current_str = input.getString(0)
    //判断最后一次的值是否包含当前的值
    if(!last_str.contains(current_str)){
      //判断是否是第一个值,是的话走if赋值,不是的话走else追加
      if(last_str.equals("")){
        last_str = current_str
      }else{
        last_str += "," + current_str
      }
    }
    buffer.update(0,last_str)

  }
/**
  *对分区结果进行合并
  * buffer1是机器hadoop1上的结果
  * buffer2是机器Hadoop2上的结果
  * */
  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    var buf1 = buffer1.getString(0)
    val buf2 = buffer2.getString(0)
    //将buf2里面存在的数据而buf1里面没有的数据追加到buf1
    //buf2的数据按照,进行切分
    for(s <- buf2.split(",")){
      if(!buf1.contains(s)){
        if(buf1.equals("")){
          buf1 = s
        }else{
          buf1 += s
        }
      }
    }
    buffer1.update(0,buf1)
  }
/**
  * 最终的计算结果
  * */
  override def evaluate(buffer: Row): Any = {
    buffer.getString(0)
  }
}

注册自定义的UDF函数为临时函数

def main(args: Array[String]): Unit = {
    /**
      * 第一步 创建程序入口
      */
    val conf = new SparkConf().setAppName("AralHotProductSpark")
    val sc = new SparkContext(conf)
    val hiveContext = new HiveContext(sc)
  //注册成为临时函数
    hiveContext.udf.register("get_distinct_city",GetDistinctCityUDF)
  //注册成为临时函数
    hiveContext.udf.register("get_product_status",(str:String) =>{
      var status = 0
      for(s <- str.split(",")){
        if(s.contains("product_status")){
          status = s.split(":")(1).toInt
        }
      }
    })
}

原文地址:https://www.cnblogs.com/cjunn/p/12234415.html

时间: 2024-10-29 08:15:28

Spark学习之路 (十九)SparkSQL的自定义函数UDF[转]的相关文章

Spark学习之路 (十九)SparkSQL的自定义函数UDF

讨论QQ:1586558083 在Spark中,也支持Hive中的自定义函数.自定义函数大致可以分为三种: UDF(User-Defined-Function),即最基本的自定义函数,类似to_char,to_date等 UDAF(User- Defined Aggregation Funcation),用户自定义聚合函数,类似在group by之后使用的sum,avg等 UDTF(User-Defined Table-Generating Functions),用户自定义生成函数,有点像str

Spark(十八)SparkSQL的自定义函数UDF

在Spark中,也支持Hive中的自定义函数.自定义函数大致可以分为三种: UDF(User-Defined-Function),即最基本的自定义函数,类似to_char,to_date等 UDAF(User- Defined Aggregation Funcation),用户自定义聚合函数,类似在group by之后使用的sum,avg等 UDTF(User-Defined Table-Generating Functions),用户自定义生成函数,有点像stream里面的flatMap 自定

Spark学习之路 (九)SparkCore的调优之数据倾斜调优

摘抄自:https://tech.meituan.com/spark-tuning-pro.html 数据倾斜调优 调优概述 有的时候,我们可能会遇到大数据计算中一个最棘手的问题——数据倾斜,此时Spark作业的性能会比期望差很多.数据倾斜调优,就是使用各种技术方案解决不同类型的数据倾斜问题,以保证Spark作业的性能. 数据倾斜发生时的现象 绝大多数task执行得都非常快,但个别task执行极慢.比如,总共有1000个task,997个task都在1分钟之内执行完了,但是剩余两三个task却要

Linux嵌入式驱动学习之路(十九)触摸屏驱动

触摸屏使用流程: 1. 按下产生中断. 2.在中断处理程序中启动AD转换XY坐标. 3.AD转换结束并产生AD中断. 4. 在AD的中断处理函数中上报信息,启动定时器. 5. 定时器时间到后进入中断,处理长按滑动.跳转到第二步 6. 松开. sd

Android学习笔记二十九之SwipeRefreshLayout、RecyclerView和CardView

Android学习笔记二十九之SwipeRefreshLayout.RecyclerView和CardView 前面我们介绍了AlertDialog和几个常用的Dialog,ProgressDialog进度条提示框.DatePickerDialog日期选择对话框和TimePickerDialog时间选择对话框.这一节我们介绍几个新的API控件SwipeRefreshLayout.RecyclerView和CardView,这几个API控件都是google在Android5.0推出的.下面我们来学

我的编程之路(十九) 开发中一些细节与启发

1.js的命名空间           如果写后台代码,分层是潜意识中的基本常识,但是一到了前台,却没了这种意识,归根结底还是js用的不多,也一直没有在意js的地位,直到现在富客户端的趋势与要求,使得很多代码都要在前台用js或其框架完成,所以对于js代码的管理就要像后台java代码一样有其规范了,而命名空间就是package,也是为了管理不同层次的代码. 2.闭包          闭包就是能够读取其他函数内部变量的函数.它的最大用处有两个,一个是可以读取函数内部的变量,另一个就是让这些变量的值

《Javascript权威指南》学习笔记之十九--HTML5 DOM新标准---处理文档元信息和管理交互能力

一.了解DOM 1.DOM是Document Object Model的缩写,即文档对象类型,是文档在内存中的表示形式,是一个应用程序接口,定义了文档的逻辑结构以及一套访问和处理文档的方法. 2.HTML DOM与Core DOM的区别:前者提供了大量的方法和属性,与现有的程序模型一致,更便于脚本的编写者控制. 二.document对象 使用window.document属性返回一个document对象,代表当前window内加载的文档.window可以省略.winName.document返回

【Unity 3D】学习笔记二十九:游戏实例——简单小地图制作

任何的学习,光看不练是学不好的.所以这次就总结回顾下怎么制作MMROPG类游戏中的小地图.在MMROPG类游戏里,主角在游戏世界里走动时,一般在屏幕右上角都会有一个区域来显示当前游戏场景的小地图.主角在游戏世界里走动,小地图里代表着主角的小标记也会随之移动.那怎么实现咧? 首先需要确定两个贴图,第一个是右上角的小地图背景贴图,应该是从Y轴俯视向下截取主角所在的位置大地图.第二个就是主角的位置大贴图.在本例中,因为没有学习unity地图制作,所以地图用一个面对象代替,主角用立方体代替,使用GUI来

Android学习路线(十九)支持不同设备——支持不同(Android)平台版本

当最新的Android版本为你的应用提供着很棒的APIs时,你却要在更多的设备更新之前继续支持老的系统版本.这篇课程如何在继续支持低版本的系统的情况下使用新版本的高级API. Platform Versions 仪表板展示了最新的活跃设备上运行的Android系统版本的分布,基于设备访问Google Play商店的次数.通常情况下,支持90%的活跃设备同时使用最新版本作为target是一个好习惯. 贴士: 为了在不同的Android版本上提供最好的特性和功能,你应该在你的应用中使用Android