spark自定義函數之——UDF使用詳解及代碼示例

前言

本文介紹如何在Spark Sql和DataFrame中使用UDF，如何利用UDF給一個表或者一個DataFrame根據需求添加幾列，並給出了舊版（Spark1.x）和新版（Spark2.x）完整的代碼示例。

關於UDF：UDF：User Defined Function，用戶自定義函數

創建測試用DataFrame

spark2.0創建DataFrame

// 構造測試數據，有兩個字段、名字和年齡
val userData = Array(("A", 16), ("B", 21), ("B", 14), ("B", 18))

//創建測試df
val userDF = spark.createDataFrame(userData).toDF("name", "age")
userDF.show

+-----+---+
| name|age|
+-----+---+
| A | 16|
| B | 21|
| C | 14|
| D | 18|
+-----+---+

// 注冊一張user表 
userDF.createOrReplaceTempView("user")

spark1.0創建DataFrame

 // 構造測試數據，有兩個字段、名字和年齡
val userData = Array(("A", 16), ("B", 21), ("C", 14), ("D", 18))
//創建測試df
val userDF = sc.parallelize(userData).toDF("name", "age")
// 注冊一張user表
 userDF.registerTempTable("user")

spark-sql中SQL中UDF用法

1. 通過匿名函數注冊UDF

下面的UDF的功能是計算某列的長度，該列的類型為String

// Spark2.x:
spark.udf.register("strLen", (str: String) => str.length())

// Spark1.x:
sqlContext.udf.register("strLen", (str: String) => str.length())

// 僅以Spark2.x為例
spark.sql("select name,strLen(name) as name_len from user").show

 

2. 通過實名函數注冊UDF

實名函數的注冊有點不同，要在后面加 _(注意前面有個空格) 

// 定義一個實名函數

/**
 * 根據年齡大小返回是否成年 成年：true,未成年：false
*/
def isAdult(age: Int) = {
  if (age < 18) {
    false
  } else {
    true
  }
}

// 注冊（僅以Spark2.x為例）
spark.udf.register("isAdult", isAdult _)

spark-sql中DataFrame中UDF用法

DataFrame的udf方法雖然和Spark Sql的名字一樣，但是屬於不同的類，它在org.apache.spark.sql.functions里，下面是它的用法

1. 注冊

import org.apache.spark.sql.functions._

//方法一：注冊自定義函數（通過匿名函數）
val strLen = udf((str: String) => str.length())

//方法二：注冊自定義函數（通過實名函數）
val udf_isAdult = udf(isAdult _)

2. 使用

可通過withColumn和select使用,下面的代碼已經實現了給user表添加兩列的功能 
* 通過看源碼，下面的withColumn和select方法Spark2.0.0之后才有的，關於spark1.xDataFrame怎么使用注冊好的UDF沒有研究

// 通過withColumn添加列
userDF.withColumn("name_len", strLen(col("name"))).withColumn("isAdult", udf_isAdult(col("age"))).show

//通過select添加列
userDF.select(col("*"), strLen(col("name")) as "name_len", udf_isAdult(col("age")) as "isAdult").show

+-----+---+--------+-------+
| name|age|name_len|isAdult|
+-----+---+--------+-------+
| A | 16| 3| false|
| B | 21| 5| true|
| C | 14| 4| false|
| D | 18| 3| true|
+-----+---+--------+-------+

withColumn和select的區別

可通過withColumn的源碼看出withColumn的功能是實現增加一列，或者替換一個已存在的列，他會先判斷DataFrame里有沒有這個列名，如果有的話就會替換掉原來的列，沒有的話就用調用select方法增加一列，所以如果我們的需求是增加一列的話，兩者實現的功能一樣，且最終都是調用select方法，但是withColumn會提前做一些判斷處理，所以withColumn的性能不如select好。 

注：select方法和sql 里的select一樣，如果新增的列名在表里已經存在，那么結果里允許出現兩列列名相同但數據不一樣，大家可以自己試一下。

 參考：https://dongkelun.com/2018/08/02/sparkUDF/