spark rdd dataset sql udf udaf

以后 为了操作的便利性， 把逻辑都包装成  udf ,udaf .  
写一个  包装接口， 对一份数据的操作  ，
直接 在 repl   给 hdfs 加上 meta  desc
在 repl  直接   写  sql .   出来的结果，直接拿 save .  
可以积累 业务逻辑。 重用






import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types._
import org.apache.spark.{SparkContext, SparkConf}
import yunzhi.utils._


object wordCount {
  def main(args: Array[String]) {


    val spark = SparkSession
      .builder()
      .appName("Spark SQL Data Soures Example")
      .config("spark.some.config.option", "some-value")
      .getOrCreate()




    val data = spark.sparkContext.textFile("testdata/c01/wc.txt") //读取文件
.flatMap(_.split(" "))
      .map((_, 10))

    data.cache()


    // Rdd
data.reduceByKey(_ + _)
      .collect()
      .foreach(println) //word计数
    // sql  rdd => DataFrame  要加入implicits 转换
import spark.implicits._

    val dataDF =   data.toDF()
    dataDF.createOrReplaceTempView("dataDF")
    spark.sql("SELECT _1,sum(_2) as cnt  FROM dataDF group by _1").show()


    // register UDF
spark.sqlContext.udf.register("strLen", (s: String) => s.length())
    spark.sql("SELECT strLen(_1),sum(_2) as cnt  FROM dataDF group by _1").show()





    //register UDAF  wordCountUDAF(String)
spark.sqlContext.udf.register("wordCountUDAF", new wordCountUDAF)

    spark.sql("SELECT strLen(_1),wordCountUDAF(_1) as cnt  FROM dataDF group by _1").show()

    //register UDAF  sum(Int)
spark.sqlContext.udf.register("IntSumUDAF", new IntSumUDAF)
    spark.sql("SELECT _1,wordCountUDAF(_1) as countcnt , IntSumUDAF(_2) as sumcnt   FROM dataDF group by _1").show()



  }
}



package yunzhi.utils

import org.apache.spark.sql.Row
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types._

/**
  * Created by l on 16-10-13.
  */
class UDAFUtil {

}



class wordCountUDAF extends UserDefinedAggregateFunction{ //ctrl+I实现复写方法
/**
    * 该方法指定具体输入数据的类型   Array 类型可以输入 多个参数,定义多个 StructField   , Array 格式 ,由于 sql 中,可能传入,多个列, 在 udaf 中 都是数组.
    * @return
*/
override def inputSchema: StructType = StructType(Array(StructField("input", StringType, true),StructField("input", StringType, true)))
  /**
    * 在进行聚合操作的时候要处理的数据的结果的类型 Array 可以定义多个 StructField  Array 格式
    * @return
*/
override def bufferSchema: StructType = StructType(Array(StructField("count", IntegerType, true)))
  /**
    * 指定UDAF函数计算后返回的结果类型
    * @return
*/
override def dataType: DataType = IntegerType
  override def deterministic: Boolean = true
/**
    * 在Aggregate之前每组数据的初始化结果
    * @param buffer
*/
override def initialize(buffer: MutableAggregationBuffer): Unit = {buffer(0)=0}
  /**
    * 在进行聚合的时候有新的值进来，对分组后的聚合如何进行计算
    * 本地的聚合操作，相当于Hadoop MapReduce模型中的Combiner(这里的Row跟DataFrame的Row无关)   worker 里面先计算
    * @param buffer
* @param input
*/
override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    buffer(0) = buffer.getAs[Int](0) + 1
}
  /**
    * 最后在分布式节点进行Local Reduce完成后需要进行全局级别的Merge操作   worker 之前的 计算
    * @param buffer1
* @param buffer2
*/
override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    buffer1(0) = buffer1.getAs[Int](0) + buffer2.getAs[Int](0)
  }
  /**
    *  返回UDAF最后的计算结果
    * @param buffer
* @return
*/
override def evaluate(buffer: Row): Any = buffer.getAs[Int](0)
}






// 和  aggrerate 函数的参数的行为 很类似   .   initialize update   merge ......
class IntSumUDAF extends UserDefinedAggregateFunction{ //ctrl+I实现复写方法
/**
    * 该方法指定具体输入数据的类型
    * @return
*/
override def inputSchema: StructType = StructType(Array(StructField("input", IntegerType, true)))
  /**
    * 在进行聚合操作的时候要处理的数据的结果的类型
    * @return
*/
override def bufferSchema: StructType = StructType(Array(StructField("count", IntegerType, true)))
  /**
    * 指定UDAF函数计算后返回的结果类型
    * @return
*/
override def dataType: DataType = IntegerType
  override def deterministic: Boolean = true
/**
    * 在Aggregate之前每组数据的初始化结果
    * @param buffer
*/
override def initialize(buffer: MutableAggregationBuffer): Unit = {buffer(0)=0}
  /**
    * 在进行聚合的时候有新的值进来，对分组后的聚合如何进行计算
    * 本地的聚合操作，相当于Hadoop MapReduce模型中的Combiner(这里的Row跟DataFrame的Row无关)   worker 里面先计算
    * @param buffer
* @param input
*/
override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    buffer(0) = buffer.getAs[Int](0) + input.getAs[Int](0)
  }
  /**
    * 最后在分布式节点进行Local Reduce完成后需要进行全局级别的Merge操作   worker 之前的 计算
    * @param buffer1
* @param buffer2
*/
override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    buffer1(0) = buffer1.getAs[Int](0) + buffer2.getAs[Int](0)
  }
  /**
    *  返回UDAF最后的计算结果
    * @param buffer
* @return
*/
override def evaluate(buffer: Row): Any = buffer.getAs[Int](0)



}