这期内容当中小编将会给大家带来有关spark2.4.3中sparkSQL用户自定义函数该怎么理解，文章内容丰富且以专业的角度为大家分析和叙述，阅读完这篇文章希望大家可以有所收获。
从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 | 设置各种配置 |
使用的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()
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(<function1>,StringType,Some(List(StringType)))
scala> spark.sql("select addName(name) as name from people").show
+------------+
| name|
+------------+
|name:Michael|
| name:Andy|
| name:Justin|
+------------+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用户自定义函数该怎么理解了，如果刚好有类似的疑惑，不妨参照上述分析进行理解。如果想知道更多相关知识，欢迎关注亿速云行业资讯频道。
亿速云「云服务器」，即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘，价格低至29元/月。点击查看>>
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
原文链接：http://blog.itpub.net/69908925/viewspace-2645049/
