Spark RDD案例：分组排行榜
【摘要】
文章目录
一、提出任务二、完成任务（一）新建Maven项目（二）添加相关依赖和构建插件（三）创建日志属性文件（四）创建分组排行榜单例对象（五）本地运行程序，查看结果（六）交互式操作查看中间结果1、...
一、提出任务
- 分组求TopN是大数据领域常见的需求，主要是根据数据的某一列进行分组，然后将分组后的每一组数据按照指定的列进行排序，最后取每一组的前N行数据。
- 有一组学生成绩数据
张三丰 90
李孟达 85
张三丰 87
王晓云 93
李孟达 65
张三丰 76
王晓云 78
李孟达 60
张三丰 94
王晓云 97
李孟达 88
张三丰 80
王晓云 88
李孟达 82
王晓云 98
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 同一个学生有多门成绩，现需要计算每个学生分数最高的前3个成绩，期望输出结果如下所示：
张三丰：94 90 87
李孟达：88 85 82
王晓云：98 97 93
- 1
- 2
- 3
- 预备工作：启动集群的HDFS与Spark
- 将成绩文件 -
grades.txt上传到HDFS上/input目录
- 实现思路：使用Spark RDD的
groupByKey()算子可以对(key, value)形式的RDD按照key进行分组，key相同的元素的value将聚合到一起，形成(key, value-list)，将value-list中的元素降序排列取前N个即可。
二、完成任务
（一）新建Maven项目
- 设置项目信息（项目名、保存位置、组编号、项目编号）
- 单击【Finish】按钮
- 将
java目录改成scala目录
（二）添加相关依赖和构建插件
- 在
pom.xml文件里添加依赖与Maven构建插件
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>net.huawei.rdd</groupId>
<artifactId>GradeTopN</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>2.11.12</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.11</artifactId>
<version>2.1.1</version>
</dependency>
</dependencies>
<build>
<sourceDirectory>src/main/scala</sourceDirectory>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>3.3.0</version>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>net.alchim31.maven</groupId>
<artifactId>scala-maven-plugin</artifactId>
<version>3.3.2</version>
<executions>
<execution>
<id>scala-compile-first</id>
<phase>process-resources</phase>
<goals>
<goal>add-source</goal>
<goal>compile</goal>
</goals>
</execution>
<execution>
<id>scala-test-compile</id>
<phase>process-test-resources</phase>
<goals>
<goal>testCompile</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
（三）创建日志属性文件
- 在资源文件夹里创建日指数型文件 -
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
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
（四）创建分组排行榜单例对象
- 在
net.huawei.rdd包里创建GradeTopN单例对象
package net.huawei.rdd
import org.apache.spark.{SparkConf, SparkContext}
/**
* 功能：成绩分组排行榜
* 作者：华卫
* 日期：2022年06月15日
*/
object GradeTopN {
def main(args: Array[String]): Unit = {
// 创建Spark配置对象
val conf = new SparkConf()
.setAppName("GradeTopN") // 设置应用名称
.setMaster("local[*]") // 设置主节点位置（目前本地调试）
// 基于Spark配置对象创建Spark容器
val sc = new SparkContext(conf);
// 实现分组排行榜
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._1
val top3 = item._2.toList.sortWith(_ > _).take(3)
(name, top3)
}) // 值排序，取前三
// 输出分组排行榜结果
top3.collect.foreach(line => {
val name = line._1
var scores = ""
for (score <- line._2)
scores = scores + " " + score
println(name + ":" + scores)
})
// 停止Spark容器，结束任务
sc.stop()
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
（五）本地运行程序，查看结果
- 在控制台查看输出结果
（六）交互式操作查看中间结果
1、读取成绩文件得到RDD
- 执行命令：
val lines = sc.textFile("hdfs://master:9000/input/grades.txt")
2、利用映射算子生成二元组构成的RDD
val grades = lines.map(line => {
val fields = line.split(" ")
(fields(0), fields(1))
})
- 1
- 2
- 3
- 4
- 执行上述代码
3、按键分组得到新的二元组构成的RDD
- 执行命令：
val groupGrades = grades.groupByKey()
4、按值排序，取前三
val top3 = groupGrades.map(item => {
val name = item._1
val top3 = item._2.toList.sortWith(_ > _).take(3)
(name, top3)
})
- 1
- 2
- 3
- 4
- 5
- 执行上述代码
5、按指定格式输出结果
top3.collect.foreach(line => {
val name = line._1
var scores = ""
for (score <- line._2)
scores = scores + " " + score
println(name + ":" + scores)
})
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 执行上述代码
文章来源: howard2005.blog.csdn.net，作者：howard2005，版权归原作者所有，如需转载，请联系作者。
原文链接：howard2005.blog.csdn.net/article/details/125289810
【版权声明】本文为华为云社区用户转载文章，如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱：
cloudbbs@huaweicloud.com
- 点赞
- 收藏
- 关注作者
评论（0）