package UDF_UDAF

import java.util

import org.apache.spark.sql.{RowFactory, SparkSession}

import org.apache.spark.SparkConf

import org.apache.spark.sql.api.java.UDF1

import org.apache.spark.sql.types.{DataTypes, StructField}

// 自定义一个继承自 UDF1（或UDF2，UDF3，UDF4...）的类

class UDF extends UDF1[String,Int]{

  override def call(t1: String): Int = {

    t1.length

  }

}

object UDF{

  def main(args: Array[String]): Unit = {

    val warehouseLocation = "/code/VersionTest/spark-warehouse" //必须是相对路径

    val conf = new SparkConf().setMaster("local").setAppName("udf")

    val sparkSession = SparkSession.builder()

      .config(conf)

      .config("spark.sql.warehouse.dir", warehouseLocation) //设置warehouse

      .getOrCreate()

    val sc = sparkSession.sparkContext

    val parallize = sc.parallelize(Array("zhangsan","lisi","wangwu"))

    val rowRDD = parallize.map(s=>RowFactory.create(s))

    val fields = new util.ArrayList[StructField]()

    fields.add(DataTypes.createStructField("name",DataTypes.StringType,true))

    val schema = DataTypes.createStructType(fields)

    val df = sqlSession.createDataFrame(rowRDD, schema)

    df.createOrReplaceTempView("user")

    sparkSession.udf.register("StrLen", new UDF(),DataTypes.IntegerType)

    sparkSession.sql("select name, StrLen(name) as length from user").show()

    sparkSession.stop()

  }

}