正文 

当前位置： 首页 > 代码 > java教程

spark学习之路（十九）sparksql的自定义函数udf(代码片段)

liuys635 liuys635     2022-12-18     179

关键词：

讨论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),用户自定义生成函数，有点像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
        
      
    )
技术图片

 

spark学习之路（十八）sparksql简单使用[转]

SparkSQL的进化之路1.0以前：Shark1.1.x开始：SparkSQL(只是测试性的)SQL1.3.x:SparkSQL(正式版本)+Dataframe1.5.x:SparkSQL钨丝计划1.6.x：SparkSQL+DataFrame+DataSet(测试版本)1.x:SparkSQL+DataFrame+DataSet(正式版本)SparkSQL:还有其他的优化StructuredS 查看详情

spark学习之路（十八）sparksql简单使用(代码片段)

讨论QQ：1586558083目录一、SparkSQL的进化之路二、认识SparkSQL2.1　什么是SparkSQL?2.2　SparkSQL的作用2.3　运行原理2.4　特点2.5　SparkSession2.7　DataFrames  三、RDD转换成为DataFrame3.1　方式一：通过 caseclass 创建 DataFrames（... 查看详情

spark学习之路（二十）sparksql的元数据

...目录一、概述二、Spark-SQL脚本 正文回到顶部一、概述SparkSQL的元数据的状态有两种：1、in_memory,用完了元数据也就丢了2、hive,通过hive去保存的，也就是说，hive的元数据存在哪儿，它的元数据也就存在哪儿。 　换句话说，... 查看详情

spark学习之路（二十）sparksql的元数据[转]

概述SparkSQL的元数据的状态有两种：1、in_memory,用完了元数据也就丢了2、hive,通过hive去保存的，也就是说，hive的元数据存在哪儿，它的元数据也就存在哪儿。换句话说，SparkSQL的数据仓库在建立在Hive之上实现的。我们要用SparkSQL... 查看详情

spark学习之路（二十三）sparkstreaming的官方文档(代码片段)

一、SparkCore、SparkSQL和SparkStreaming的类似之处二、SparkStreaming的运行流程2.1　图解说明 2.2　文字解说1、我们在集群中的其中一台机器上提交我们的ApplicationJar，然后就会产生一个Application，开启一个Driver，然后初始化SparkStreamin... 查看详情

sparksql自定义函数(代码片段)

目录1SparkSQL自定义函数1.1自定义函数分类1.2自定义UDF1.3自定义UDAF2开窗函数2.1、概述2.2.准备工作2.3.聚合开窗函数2.4.排序开窗函数2.4.1ROW_NUMBER顺序排序2.4.2RANK跳跃排序2.4.3DENSE_RANK连续排序2.4.4NTILE分组排名1SparkSQL自定义函数1.1自定... 查看详情

：sparksql函数定义

Spark是大数据体系的明星产品，是一款高性能的分布式内存迭代计算框架，可以处理海量规模的数据。下面就带大家来学习今天的内容！ 往期内容：Spark基础入门-第一章：Spark框架概述Spark基础入门-第二章ÿ... 查看详情

spark学习之路spark的广播变量和累加器(代码片段)

讨论QQ：1586558083目录一、概述二、广播变量broadcastvariable2.1　为什么要将变量定义成广播变量？2.2　广播变量图解2.3　如何定义一个广播变量？2.4　如何还原一个广播变量？2.5　定义广播变量需要的注意点？2.6　注意事项三、累... 查看详情

sparksql如何自定义函数

 1.SparkSql如何自定义函数2.示例：Average3.类型安全的自定义函数1.SparkSql如何自定义函数？　　spark中我们定义一个函数，需要继承UserDefinedAggregateFunction这个抽象类，实现这个抽象类中所定义的方法，这是一个模板设计模式？... 查看详情

Spark SQL - 自定义数据类型 UUID

】SparkSQL-自定义数据类型UUID【英文标题】：SparkSQL-CustomDatatypeUUID【发布时间】：2018-04-0308:26:02【问题描述】：我正在尝试使用SparkSQL中的自定义数据类型将数据集中的列从varchar转换为UUID。但我看到转换没有发生。如果我在这里... 查看详情

apache spark中的自定义分区器

...espark【发布时间】：2016-02-0220:33:41【问题描述】：我正在学习《学习火花：闪电般的大数据分析》一书中的示例：//custompartitionerclassDomainNamePartitioner(numParts:Int)extendsPartitioneroverridedefnumPartitions:I 查看详情

spark学习之sparksql语法优化

...park的资源调度进行了学习，今天我们要学习的内容是sparksql语法优化部分，对往期内容感兴趣的同学可以参考👇:上一篇:spark学习之资源调度.上一篇:spark学习之执行计划explain.hadoop专题:hadoop系列文章.spark专题:spark系列... 查看详情

学习笔记spark——sparksql应用——sparksql简介环境配置(代码片段)

一、SparkSQL简介1.1、SparkSQL特性SparkSQL是SparkCore之上的一个组件，它引入了一个称为SchemaRDD的新-数据抽象，它为结构化和半结构化数据提供支持提供了DataFrame、DataSet的编程抽象可以充当分布式SQL查询引擎SparkSQL是spark套件... 查看详情

spark学习之sparksql语法优化(代码片段)

...park的资源调度进行了学习，今天我们要学习的内容是sparksql语法优化部分，对往期内容感兴趣的同学可以参考👇:上一篇:spark学习之资源调度.上一篇:spark学习之执行计划explain.hadoop专题:hadoop系列文章.spark专题:spark系列... 查看详情

spark学习之路sparkcore的调优之shuffle调优

讨论QQ：1586558083目录一、概述二、shuffle的定义三、ShuffleManager发展概述四、HashShuffleManager的运行原理4.1　未经优化的HashShuffleManager4.2　优化后的HashShuffleManager五、SortShuffleManager运行原理5.1　普通运行机制5.2　bypass运行机制六、shu... 查看详情

hive学习之路（十九）hive的数据倾斜(代码片段)

1、什么是数据倾斜？由于数据分布不均匀，造成数据大量的集中到一点，造成数据热点2、Hadoop框架的特性　　A、不怕数据大，怕数据倾斜　　B、Jobs数比较多的作业运行效率相对比较低，如子查询比较多　　C、sum,count,max,min等... 查看详情

spark成长之路(13)-dataset与dataframe(代码片段)

...e总结前言spark1.6中引入了DataSet和DataFrame的概念，然后SparkSQL的API也是基于这两个概念的，到2.2推出的稳定版本的StructuredStreaming也是依靠SparkSQL 查看详情

spark学习之路（二十三）sparkstreaming的官方文档[转](代码片段)

SparkCore、SparkSQL和SparkStreaming的类似之处SparkStreaming的运行流程1、我们在集群中的其中一台机器上提交我们的ApplicationJar，然后就会产生一个Application，开启一个Driver，然后初始化SparkStreaming的程序入口StreamingContext；2、Master会为这... 查看详情