spark自定义函数之——UDF使用详解及代码示例

// 构造测试数据，有两个字段、名字和年龄
val userData = Array(("A", ), ("B", ), ("B", ), ("B", ))//创建测试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")

// 构造测试数据，有两个字段、名字和年龄
val userData = Array(("A", ), ("B", ), ("C", ), ("D", ))
//创建测试df
val userDF = sc.parallelize(userData).toDF("name", "age")
// 注册一张user表
userDF.registerTempTable("user")

// 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

// 定义一个实名函数/**
* 根据年龄大小返回是否成年 成年：true,未成年：false
*/
def isAdult(age: Int) = {
if (age < ) {
false
} else {
true
}
}// 注册（仅以Spark2.x为例）
spark.udf.register("isAdult", isAdult _)

import org.apache.spark.sql.functions._
//方法一：注册自定义函数（通过匿名函数）
val strLen = udf((str: String) => str.length())//方法二：注册自定义函数（通过实名函数）
val udf_isAdult = udf(isAdult _)

// 通过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 | | | false|
| B | | | true|
| C | | | false|
| D | | | true|
+-----+---+--------+-------+