Spark實現分組TopN

一.概述

　　在許多數據中，都存在類別的數據，在一些功能中需要根據類別分別獲取前幾或后幾的數據，用於數據可視化或異常數據預警。在這種情況下，實現分組TopN就顯得非常重要了，因此，使用了Spark聚合函數和排序算法實現了分布式TopN計算功能。

　　

二.代碼實現

package scala

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{Row, SparkSession}

/**
  * 計算分組topN
  * Created by Administrator on 2019/11/20.
  */
object GroupTopN {
  Logger.getLogger("org").setLevel(Level.WARN) // 設置日志級別
  def main(args: Array[String]) {
    //創建測試數據
    val test_data = Array("CJ20191120,201911", "CJ20191120,201910", "CJ20191105,201910", "CJ20191105,201909", "CJ20191111,201910")
    val spark = SparkSession.builder().appName("GroupTopN").master("local[2]").getOrCreate()
    val sc = spark.sparkContext
    val test_data_rdd = sc.parallelize(test_data).map(row => {
      val Array(scene, cycle) = row.split(",")
      Row(scene, cycle)
    })
    // 設置數據模式
    val structType = StructType(Array(
      StructField("scene", StringType, true),
      StructField("cycle", StringType, true)
    ))
    // 轉換為df
    val test_data_df = spark.createDataFrame(test_data_rdd, structType)
    test_data_df.createOrReplaceTempView("test_data_df")
    // 拼接周期
    val scene_ws = spark.sql("select scene,concat_ws(',',collect_set(cycle)) as cycles from test_data_df group by scene")
    scene_ws.count()
    scene_ws.show()
    scene_ws.createOrReplaceTempView("scene_ws")
    /**
      * 定義參數確定N的大小，暫定為1
      */
    val sum = 1
    // 創建廣播變量，把N的大小廣播出去
    val broadcast = sc.broadcast(sum)
    /**
      * 定義Udf實現獲取組內的前N個數據
      */
    spark.udf.register("getTopN", (cycles : String) => {
      val sum = broadcast.value
      var mid = ""
      if(cycles.contains(",")){ // 多值
        val cycle = cycles.split(",").sorted.reverse // 降序排序
        val min = Math.min(cycle.length, sum)
        for(i <- 0 until min){
          if(mid.equals("")){
            mid = cycle(i)
          }else{
            mid += "," + cycle(i)
          }
        }
      }else{ // 單值
        mid = cycles
      }
      mid
    })

    val result = spark.sql("select scene,getTopN(cycles) cycles from scene_ws")
    result.show()
    spark.stop()
  }
}

三.結果

　　

　　

四.備注

　　當N大於1時，多個數據會拼接在一起，若想每個一行，可是使用使用列轉行功能，參考我的博客：https://www.cnblogs.com/yszd/p/11266552.html