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()
}
}