spark2.4.3中sparkSQL用戶自定義函數該怎么理解

這期內容當中小編將會給大家帶來有關spark2.4.3中sparkSQL用戶自定義函數該怎么理解，文章內容豐富且以專業的角度為大家分析和敘述，閱讀完這篇文章希望大家可以有所收獲。

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 的方法如下：

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(<function1>,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用戶自定義函數該怎么理解了，如果剛好有類似的疑惑，不妨參照上述分析進行理解。如果想知道更多相關知識，歡迎關注億速云行業資訊頻道。