Spark

import java.util

import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory
import org.apache.hadoop.hive.serde2.objectinspector.{ObjectInspector, ObjectInspectorFactory, StructObjectInspector}
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataframe, SparkSession}


//一进多出
object UDTF {
  def main(args: Array[String]): Unit = {
    val sparkSession: SparkSession = SparkSession.builder().appName("UDTF").master("local[*]")
      .config("hive.metastore.uris", "thrift://192.168.153.101:9083")
      .enableHiveSupport()
      .getOrCreate()

    val sc: SparkContext = sparkSession.sparkContext

    import sparkSession.implicits._

    val rdd: RDD[String] = sc.textFile("in/UDTF.txt")
    val rdd2: RDD[(String, String, String)] = rdd
      .map(x => x.split("//"))
      .filter(x => x(1).equals("ls"))
      .map(x => (x(0),x(1),x(2)))

    val frame: Dataframe = rdd2.toDF("id","name","class")
    frame.printSchema()
    frame.show(false)

    frame.createOrReplaceTempView("udtfTable")

    sparkSession.sql("create temporary function Myudtf as 'kb15.function.MyUDTF'")
    sparkSession.sql("select Myudtf(class) from udtfTable").show(false)
  }
}
class MyUDTF extends GenericUDTF{

  override def initialize(argOIs: Array[ObjectInspector]): StructObjectInspector = {
    val fieldName: util.ArrayList[String] = new java.util.ArrayList[String]()
    val fieldOIs: util.ArrayList[ObjectInspector] = new java.util.ArrayList[ObjectInspector]()

    //定义输出字段类型
    fieldName.add("type")
    fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector)
    ObjectInspectorFactory.getStandardStructObjectInspector(fieldName,fieldOIs)
  }

  

  override def process(objects: Array[AnyRef]): Unit = {
    val strings: Array[String] = objects(0).toString.split(" ")
    for (str<- strings){
      val temp: Array[String] = new Array[String](1)
      temp(0) = str
      forward(temp)
    }
  }

  override def close(): Unit = {

  }
}