一、提出任務(wù)

• 分組求TopN是大數(shù)據(jù)領(lǐng)域常見的需求，主要是根據(jù)數(shù)據(jù)的某一列進(jìn)行分組，然后將分組后的每一組數(shù)據(jù)按照指定的列進(jìn)行排序，最后取每一組的前N行數(shù)據(jù)。
• 有一組學(xué)生成績數(shù)據(jù)
• 張三豐 90 李孟達(dá) 85 張三豐 87 王曉云 93 李孟達(dá) 65 張三豐 76 王曉云 78 李孟達(dá) 60 張三豐 94 王曉云 97 李孟達(dá) 88 張三豐 80 王曉云 88 李孟達(dá) 82 王曉云 98

  同一個(gè)學(xué)生有多門成績，現(xiàn)需要計(jì)算每個(gè)學(xué)生分?jǐn)?shù)最高的前3個(gè)成績，期望輸出結(jié)果如下所示：

• 張三豐：94 90 87 李孟達(dá)：88 85 82 王曉云：98 97 93

  預(yù)備工作：啟動集群的HDFS與Spark

• ?將成績文件 - grades.txt上傳到HDFS上/input目錄

• 二、完成任務(wù)

  （一）新建Maven項(xiàng)目

• 設(shè)置項(xiàng)目信息（項(xiàng)目名、保存位置、組編號、項(xiàng)目編號）

• ?

  ?將java目錄改成scala目錄

• （二）添加相關(guān)依賴和構(gòu)建插件

• 在pom.xml文件里添加依賴與Maven構(gòu)建插件

設(shè)置maven

?完成后點(diǎn)擊刷新

• （三）創(chuàng)建日志屬性文件

• 在資源文件夾里創(chuàng)建日志屬性文件 - log4j.properties

?

log4j.rootLogger=ERROR, stdout, logfile log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n log4j.appender.logfile=org.apache.log4j.FileAppender log4j.appender.logfile.File=target/spark.log log4j.appender.logfile.layout=org.apache.log4j.PatternLayout log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

（四）創(chuàng)建分組排行榜單例對象

• 在net.huawei.rdd包里創(chuàng)建GradeTopN單例對象

package net.zhj.rddimport org.apache.spark.{SparkConf, SparkContext}object GradeTopN {def main(args: Array[String]): Unit = {// 創(chuàng)建Spark配置對象val conf = new SparkConf().setAppName("GradeTopN") // 設(shè)置應(yīng)用名稱.setMaster("local[*]") // 設(shè)置主節(jié)點(diǎn)位置（目前本地調(diào)試）// 基于Spark配置對象創(chuàng)建Spark容器val sc = new SparkContext(conf);// 實(shí)現(xiàn)分組排行榜val top3 = sc.textFile("hdfs://master:9000/input/grades.txt").map(line => {val fields = line.split(" ")(fields(0), fields(1))}) // 將每行成績映射成二元組(name, grade).groupByKey() // 按鍵分組.map(item => {val name = item._1val top3 = item._2.toList.sortWith(_ > _).take(3)(name, top3)}) // 值排序，取前三// 輸出分組排行榜結(jié)果 top3.collect.foreach(line => {val name = line._1var scores = ""for (score <- line._2)scores = scores + " " + scoreprintln(name + ":" + scores)})// 停止Spark容器，結(jié)束任務(wù)sc.stop()} }

（五）本地運(yùn)行程序，查看結(jié)果

• 在控制臺查看輸出結(jié)果

（六）交互式操作查看中間結(jié)果

1、讀取成績文件得到RDD

• 執(zhí)行命令：val lines = sc.textFile("hdfs://master:9000/input/grades.txt")

?2、利用映射算子生成二元組構(gòu)成的RDD

?運(yùn)行代碼

3、按鍵分組得到新的二元組構(gòu)成的RDD

• 執(zhí)行命令：val groupGrades = grades.groupByKey()

4、按值排序，取前三

val top3 = groupGrades.map(item => {val name = item._1val top3 = item._2.toList.sortWith(_ > _).take(3)(name, top3)})

?5、按指定格式輸出結(jié)果

top3.collect.foreach(line => {val name = line._1var scores = ""for (score <- line._2)scores = scores + " " + scoreprintln(name + ":" + scores)})

執(zhí)行上述代碼

總結(jié)

以上是生活随笔為你收集整理的Spark SQL案例：分组排行榜的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。