spark2.4.3中sparkSQL用户自定义函数该怎么理解
这期内容当中小编将会给大家带来有关spark2.4.3中sparkSQL用户自定义函数该怎么理解,文章内容丰富且以专业的角度为大家分析和叙述,阅读完这篇文章希望大家可以有所收获。
创新互联网站建设提供从项目策划、软件开发,软件安全维护、网站优化(SEO)、网站分析、效果评估等整套的建站服务,主营业务为成都做网站、网站制作,成都App定制开发以传统方式定制建设网站,并提供域名空间备案等一条龙服务,秉承以专业、用心的态度为用户提供真诚的服务。创新互联深信只要达到每一位用户的要求,就会得到认可,从而选择与我们长期合作。这样,我们也可以走得更远!
1、简介
从Spark2.0以上的版本开始,spark是使用全新的SparkSession接口代替Spark1.6中的SQLcontext和HiveContext
来实现对数据的加载、转换、处理等工作,并且实现了SQLcontext和HiveContext的所有功能。
我们在新版本中并不需要之前那么繁琐的创建很多对象,只需要创建一个SparkSession对象即可。
SparkSession支持从不同的数据源加载数据,并把数据转换成DataFrame,并支持把DataFrame转换成SQLContext自身中的表。
然后使用SQL语句来操作数据,也提供了HiveQL以及其他依赖于Hive的功能支持。
创建SparkSession
SparkSession 是 Spark SQL 的入口。
使用 Dataset 或者 Datafram 编写 Spark SQL 应用的时候,第一个要创建的对象就是 SparkSession。
Builder 是 SparkSession 的构造器。 通过 Builder, 可以添加各种配置。
Builder 的方法如下:
Method | Description |
---|---|
getOrCreate | 获取或者新建一个 sparkSession |
enableHiveSupport | 增加支持 hive Support |
appName | 设置 application 的名字 |
config | 设置各种配置 |
2、sparkSQL基本使用方法
使用的spark版本2.4.3
spark 1.x中的SQLContext在新版本中已经被废弃,改为SparkSession.builder
可以写成
val conf = new SparkConf().setAppName("helloworld").setMaster("local[*]") val spark1=SparkSession.builder().config(conf).getOrCreate()
或(sparksession构造器私有化在builder中)
val spark = SparkSession.builder .appName("my spark application") .master("local[2]") .getOrCreate()
例:
import org.apache.spark.sql.SparkSession object HelloWorld { def main(args: Array[String]): Unit = { /* val conf = new SparkConf().setAppName("helloworld").setMaster("local[*]") val spark1=SparkSession.builder().config(conf).getOrCreate()*/ val spark = SparkSession.builder .appName("my spark application") .master("local[2]") .getOrCreate() //读取数据 val df = spark.read.json("/usr/local/opt/spark-2.4.3/examples/src/main/resources/people.json") //展示所有数据 df.show() //DSL df.select("name").show() //SQL df.createTempView("people") spark.sql("select * from people where age=30").show() //关闭 spark.close() } }
输出结果 1:
//展示所有数据 df.show()
输出结果 2:
//DSL df.select("name").show()
输出结果 3:
//SQL df.createTempView("people") spark.sql("select * from people where age=30").show()
3、通过udf自定义用户函数addName (实现将字段x前拼接上name:x)
scala> spark.read.json("./examples/src/main/resources/people.json") res32: org.apache.spark.sql.DataFrame = [age: bigint, name: string] scala> res32.createOrReplaceTempView("people") scala> spark.sql("select * from people") res38: org.apache.spark.sql.DataFrame = [age: bigint, name: string] scala> spark.sql("select * from people").show +----+-------+ | age| name| +----+-------+ |null|Michael| | 30| Andy| | 19| Justin| +----+-------+ scala> spark.udf.register("addName",(x:String)=> "name:"+x) res40: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(,StringType,Some(List(StringType))) scala> spark.sql("select addName(name) as name from people").show +------------+ | name| +------------+ |name:Michael| | name:Andy| | name:Justin| +------------+
4、通过udaf自定义用户函数
package com.ny.service import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction} import org.apache.spark.sql.types._ import org.apache.spark.sql.{Row, SparkSession} class CustomerAvg extends UserDefinedAggregateFunction { //输入的类型 override def inputSchema: StructType = StructType(StructField("salary", LongType) :: Nil) //缓存数据的类型 override def bufferSchema: StructType = { StructType(StructField("sum", LongType) :: StructField("count", LongType) :: Nil) } //返回值类型 override def dataType: DataType = LongType //幂等性 override def deterministic: Boolean = true //初始化 override def initialize(buffer: MutableAggregationBuffer): Unit = { buffer(0) = 0L buffer(1) = 0L } //更新 分区内操作 override def update(buffer: MutableAggregationBuffer, input: Row): Unit = { buffer(0)=buffer.getLong(0) +input.getLong(0) buffer(1)=buffer.getLong(1)+1L } //合并 分区与分区之间操作 override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = { buffer1(0)=buffer1.getLong(0)+buffer2.getLong(0) buffer1(1)=buffer1.getLong(1)+buffer2.getLong(1) } //最终执行的方法 override def evaluate(buffer: Row): Any = { buffer.getLong(0)/buffer.getLong(1) } } object CustomerAvg{ def main(args: Array[String]): Unit = { val spark= SparkSession.builder() .appName("MyAvg") .master("local[2]") .getOrCreate() spark.udf.register("MyAvg",new CustomerAvg) //读数据 val frame = spark.read.json("/usr/local/opt/spark-2.4.3/examples/src/main/resources/peopleCP.json") frame.createTempView("peopleCP") spark.sql("select MyAvg(age) avg_age from peopleCP").show() spark.stop() } }
nancylulululu:resources nancy$ vi peopleCP.json {"name":"Michael","age":11} {"name":"Andy", "age":30} {"name":"Justin", "age":19}
返回结果
上述就是小编为大家分享的spark2.4.3中sparkSQL用户自定义函数该怎么理解了,如果刚好有类似的疑惑,不妨参照上述分析进行理解。如果想知道更多相关知识,欢迎关注创新互联行业资讯频道。
分享名称:spark2.4.3中sparkSQL用户自定义函数该怎么理解
网站网址:http://scyanting.com/article/gepcho.html