Spark基於自定義聚合函數實現【列轉行、行轉列】

一.分析

　　Spark提供了非常豐富的算子，可以實現大部分的邏輯處理，例如，要實現行轉列，可以用hiveContext中支持的concat_ws(',', collect_set('字段'))實現。但是這有明顯的局限性【sqlContext不支持】，因此，基於編碼邏輯或自定義聚合函數實現相同的邏輯就顯得非常重要了。

二.列轉行代碼實現　

package utils
import com.hankcs.hanlp.tokenizer.StandardTokenizer
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{SparkSession, Row}
import org.apache.spark.sql.types.{StringType, StructType, StructField}
/**
  * Created by Administrator on 2019/12/17.
  */
object Column2Row {
  /**
    * 設置日志級別
    */
  Logger.getLogger("org").setLevel(Level.WARN)
  def main(args: Array[String]) {
    val spark = SparkSession.builder().master("local[2]").appName(s"${this.getClass.getSimpleName}").getOrCreate()
    val sc = spark.sparkContext
    val sqlContext = spark.sqlContext

    val array : Array[String] = Array("spark-高性能大數據解決方案", "spark-機器學習圖計算", "solr-搜索引擎應用廣泛", "solr-ES靈活高效")
    val rdd = sc.parallelize(array)

    val termRdd = rdd.map(row => { // 標准分詞，掛載Hanlp分詞器
    var result = ""
      val type_content = row.split("-")
      val termList = StandardTokenizer.segment(type_content(1))
      for(i <- 0 until termList.size()){
        val term = termList.get(i)
        if(!term.nature.name.contains("w") && !term.nature.name().contains("u") && !term.nature.name().contains("m")){
          if(term.word.length > 1){
            result += term.word + " "
          }
        }
      }
      Row(type_content(0),result)
    })

    val structType = StructType(Array(
      StructField("arth_type", StringType, true),
      StructField("content", StringType, true)
    ))

    val termDF = sqlContext.createDataFrame(termRdd,structType)
    termDF.show(false)
    /**
      * 列轉行
      */
    val termCheckDF = termDF.rdd.flatMap(row =>{
      val arth_type = row.getAs[String]("arth_type")
      val content = row.getAs[String]("content")
      var res = Seq[Row]()
      val content_array = content.split(" ")
      for(con <- content_array){
        res = res :+ Row(arth_type,con)
      }
      res
    }).collect()

    val termListDF = sqlContext.createDataFrame(sc.parallelize(termCheckDF), structType)
    termListDF.show(false)

    sc.stop()
  }
}

三.列轉行執行結果

　　列轉行之前：

　　

　　列轉行：

　　

四.行轉列代碼實現

package test

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types._
import org.apache.spark.sql.{Row, SparkSession}

/**
  * 自定義聚合函數實現行轉列
  */
object AverageUserDefinedAggregateFunction extends UserDefinedAggregateFunction{
  //聚合函數輸入數據結構
  override def inputSchema:StructType = StructType(StructField("input", StringType) :: Nil)

  //緩存區數據結構
  override def bufferSchema: StructType = StructType(StructField("result", StringType) :: Nil)

  //結果數據結構
  override def dataType : DataType = StringType

  // 是否具有唯一性
  override def deterministic : Boolean = true

  //初始化
  override def initialize(buffer : MutableAggregationBuffer) : Unit = {
    buffer(0) = ""
  }

  //數據處理 : 必寫，其它方法可選，使用默認
  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    if(input.isNullAt(0)) return
    if(buffer.getString(0) == null || buffer.getString(0).equals("")){
      buffer(0) = input.getString(0) //拼接字符串
    }else{
      buffer(0) = buffer.getString(0) + "," + input.getString(0) //拼接字符串
    }
  }

  //合並
  override def merge(bufferLeft: MutableAggregationBuffer, bufferRight: Row): Unit ={
    if(bufferLeft(0) == null || bufferLeft(0).equals("")){
      bufferLeft(0) = bufferRight.getString(0) //拼接字符串
    }else{
      bufferLeft(0) = bufferLeft(0) + "," + bufferRight.getString(0) //拼接字符串
    }
  }

  //計算結果
  override def evaluate(buffer: Row): Any  = buffer.getString(0)
}

/**
  * Created by Administrator on 2019/12/17.
  */
object Row2Columns {
  /**
    * 設置日志級別
    */
  Logger.getLogger("org").setLevel(Level.WARN)
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().master("local[2]").appName(s"${this.getClass.getSimpleName}").getOrCreate()
    val sc = spark.sparkContext
    val sqlContext = spark.sqlContext

    val array : Array[String] = Array("大數據-Spark","大數據-Hadoop","大數據-Flink","搜索引擎-Solr","搜索引擎-ES")

    val termRdd = sc.parallelize(array).map(row => { // 標准分詞，掛載Hanlp分詞器
      val content = row.split("-")
      Row(content(0), content(1))
    })

    val structType = StructType(Array(
      StructField("arth_type", StringType, true),
      StructField("content", StringType, true)
    ))

    val termDF = sqlContext.createDataFrame(termRdd,structType)
    termDF.show()
    termDF.createOrReplaceTempView("term")

    /**
      * 注冊udaf
      */
    spark.udf.register("concat_ws", AverageUserDefinedAggregateFunction)
    spark.sql("select arth_type,concat_ws(content) content from term group by arth_type").show()
  }
}

五.行轉列執行結果

　　行轉列之前：

　　

　　行轉列：