SparkSQL的自定义函数UDF使用

import org.apache.spark.sql.Row
import org.apache.spark.rdd._
import scala.collection.mutable.ArrayBuffer
import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType}
// 通过RDD创建测试数据
val rdd: RDD[Row] = sc.parallelize(List("Michael,male, 29",            
"Andy,female, 30",
"Justin,male, 19",
"Dela,female, 25",
"Magi,male, 20",
"Pule,male,21"))
.map(_.split(",")).map(p => Row(p(0),p(1),p(2).trim.toInt))
// 创建Schema
val schema = StructType( Array( StructField("name",StringType, true),StructField("sex",StringType, true),StructField("age",IntegerType,true)))
// 转换DataFrame  
val peopleDF = spark.sqlContext.createDataFrame(rdd,schema) 
// 注册UDF函数    
spark.udf.reGISter("strlen",(x:String)=>x.length)
// 创建临时表       
peopleDF.registerTempTable("people")                  
// 选择输出语句，（选择输出列：名字，名字长度，性别从表people中）
spark.sql("select name, strlen(name) as strlen,sex from people").show()

scala> val df = spark.read.JSON("data/user.json")
df: org.apache.spark.sql.DataFrame = [age: bigint， username: string]

scala> spark.udf.register("addName",(x:String)=> "Name:"+x)
res9: org.apache.spark.sql.expressions.UserDefinedFunction = 
UserDefinedFunction(<function1>,StringType,Some(List(StringType)))

scala> df.createOrReplaceTempView("people")

scala> spark.sql("Select addName(name),age from people").show()