Spark学习之路(八)：分别使用Java与Scala实现Spark二次排序

内容简介

一、Spark二次排序的概念
二、实现二次排序的详细步骤(Java语言)
三、二次排序代码演示
- 1.Java版本
- 2.Scala版本
四、总结

一、Spark二次排序的概念

排序操作是数据处理过程中的常用操作步骤，Spark提供了诸如sortBy算子和sortByKey算子来实现排序，但是存在一些不足的地方，此类算子只能针对一个排序的依据进行排序，比如说sortByKey算子仅仅只能根据Key来排序，而Key常常只是一个数字或者字符串，现给出这样的一个需求(以下的演示也是围绕这个需求)，给出学生数据，里面有数学和语文两门成绩，按数学成绩升序排序，如果数学成绩相同，再按语文成绩升序排序，这里涉及到了同时对两个排序字段进行排序，因而直接使用以上的那些算子是不能实现的，而实际生产中排序字段只会更多甚至10几个，因此就要使用Spark核心编程的较为高阶的技术二次排序了。
Spark二次排序本质上是自定义Key来实现排序，使用一个Bean将所有的要排序的字段封装进Bean中，然后在Bean自定义排序规则，最后结合sortBy算子或者sortByKey算子操作自定义的Key即可实现Spark的二次排序。

二、实现二次排序的详细步骤(Java语言)

创建一个JavaBean，将涉及排序的字段封装进JavaBean，中比如说上述需求的数学成绩与语文成绩就是待排序字段，然后为排序字段增添getter和setter方法。
实现Java序列化接口Serializable，因为要被Spark直接引用所以自定义的引用类型都必须序列化，而Spark支持Java的序列化。
实现Ordered接口，其泛型类型就是当前JavaBean。然后重写接口方法：
(1).大于方法：$greater

(2).大于或等于方法：$greater$eq

(3).小于方法：$less

(4).小于或等于方法：$less$eq

(5).重写compare和compareTo方法

实际上，上述5类方法涵盖了两个待比较Key的所有比较结果，Spark会依次调用上述方法，直到产生一个比较结果就会跳出比较。
将自定义的JavaBean作为Key使用sorkByKey算子进行排序，sorkByKey算子会根据我们的自定义排序规则对Key进行排序。

三、二次排序代码演示

1.Java版本

创建类SecondlySortKeyJava，用于封装待排序的字段，并实现Order接口，实现接口方法：

/**
* 定义二次排序的key，即实现Ordered接口
* 另外因为这个Key需要被算子引用所以需要序列化
* 常见的方法是实现Java的Serializable
* 或者使用Kryo序列化
*/
public class SecondlySortKeyJava implements Ordered<SecondlySortKeyJava>, Serializable {private int math;private int chinese;// 定义有参构造方法public SecondlySortKeyJava(int manth, int chinese) {this.math = manth;this.chinese = chinese;}// 定义相等规则public int compare(SecondlySortKeyJava other) {if(this.math - other.math != 0)return this.math - other.math;else if(this.chinese - other.chinese != 0){return this.chinese - other.chinese;}return 0;}// 定义小于规则public boolean $less(SecondlySortKeyJava other) {if(this.math < other.getManth()){return true;}else if(this.math == other.getManth()){return this.chinese < other.getChinese();}return false;}// 定义大于规则public boolean $greater(SecondlySortKeyJava other) {if(this.math > other.getManth()){return true;}else if(this.math == other.math){return this.chinese > other.getChinese();}return false;}// 定义小于或等于规则public boolean $less$eq(SecondlySortKeyJava other) {if($less(other)){return true;}else if(this.math == other.getManth() && this.chinese == other.getChinese()){return true;}return false;}// 定义大于或等于规则public boolean $greater$eq(SecondlySortKeyJava other) {if($greater(other)){return true;}else if(this.math == other.getManth() && this.chinese == other.getChinese()){return true;}return false;}// 定义相等比较规则public int compareTo(SecondlySortKeyJava other) {if(this.math - other.math != 0)return this.math - other.math;else if(this.chinese - other.chinese != 0){return this.chinese - other.chinese;}return 0;}public int getManth() {return math;}public void setManth(int manth) {this.math = manth;}public int getChinese() {return chinese;}public void setChinese(int chinese) {this.chinese = chinese;}
}

编写Spark程序主类

/**
* 使用Saprk进行二次排序
*/
public class SparkSecondlySortJava {public static void main(String[] args) {SparkConf conf = new SparkConf().setMaster("local").setAppName("SparkSecondlySortJava");JavaSparkContext sc = new JavaSparkContext(conf);// 生成模拟数据List<String> keyList = new ArrayList<String>();keyList.add("75,100");keyList.add("78,65");keyList.add("85,55");keyList.add("85,95");keyList.add("95,32");keyList.add("75,77");// 序列化数据生成RDD// 将RDD映射为元组，将key切割封装进自定义的key// 降序排序后将结果输出sc.parallelize(keyList).mapToPair(new PairFunction<String, SecondlySortKeyJava, String>() {public Tuple2<SecondlySortKeyJava, String> call(String key) throws Exception {String[] keySplited = key.split(",");// 将待排序字段封装进JavaBean中return new Tuple2<SecondlySortKeyJava, String>(new SecondlySortKeyJava(Integer.valueOf(keySplited[0]),Integer.valueOf(keySplited[1])),key);}})// 排序，false代表降序排序.sortByKey(false).foreach(new VoidFunction<Tuple2<SecondlySortKeyJava, String>>() {public void call(Tuple2<SecondlySortKeyJava, String> tuple) throws Exception {System.out.println(tuple._2);}});}
}

2.Scala版本

创建类SecondlySortKeyJava，用于封装待排序的字段，并实现Order接口，实现接口方法，接口方法相比于Java版本的更为简洁：

/*** 自定义二次排序KeyScala版本* @param math* @param chinese*/
class SecondlySortKeyScala(mathScore:Int,chineseScore:Int)extends Ordered[SecondlySortKeyScala] with Serializable {// 封装排序字段var math = mathScorevar chinese = chineseScore// 重写compare方法实现比较规则override def compare(other: SecondlySortKeyScala): Int = {if(this.math != other.math){return this.math - other.math;}else if(this.chinese != other.chinese){return this.chinese - other.chinese}return 0}
}

Spark程序主类

object SparkSecondlySortScala {def main(args: Array[String]): Unit = {val conf = new SparkConf().setMaster("local").setAppName("SparkSecondlySortScala").set("spark.testing.memory", "2147480000")val sc = new SparkContext(conf)// 生成模拟数据sc.parallelize(Array("75,100", "78,65", "85,55", "85,95", "95,32", "75,77")).map(key => {val keySplited = key.split(",")Tuple2(new SecondlySortKeyScala(keySplited(0).toInt,keySplited(1).toInt),key)}).sortByKey(false).foreach(tuple => println(tuple._2))}
}

四、总结

Spark的二次排序是Spark的一个非常之常用和重要的操作，经常用于多维度的排序，比如从商品点击次数、商品的下单次数、商品的收藏次数、商品的好评度等等多维度来给予商品热门排序等。感谢您的阅读，如有错误请不吝赐教！
更多内容请查看萧邦主的技术博客导航

Spark学习之路(八)：分别使用Java与Scala实现Spark二次排序相关推荐

Spark学习之路一——Spark基础及环境搭建
Spark学习之路一--Spark基础及环境搭建文章目录一. Spark 概述 1.1 概述 1.2 优势特性 1.2.1 运行速度快 1.2.2 容易使用 1.2.3 通用性 1.2.4 运行模 ...
hadoop学习笔记（八）：Java HDFS API
hadoop学习笔记(八):Java HDFS API 阅读目录一.使用HDFS FileSystem详解回到顶部一.使用HDFS FileSystem详解 HDFS依赖的第三方包: hadoo ...
Spark学习之路（二十二）SparkStreaming的官方文档
讨论QQ:1586558083 目录一.简介 1.1 概述 1.2 一个小栗子 2.2 初始化StreamingContext 2.3 离散数据流 (DStreams) 2.4 输入DStream和 ...
Spark学习之路（十五）SparkCore的源码解读（一）启动脚本
讨论QQ:1586558083 目录一.启动脚本分析 1.1 start-all.sh 1.2 start-master.sh 1.3 spark-config.sh(1.2的第5步) 1.4 lo ...
Spark学习之路——8.Spark MLlib
MLlib的官网文档: http://spark.apache.org/docs/latest/ml-guide.html 本节主要内容: 一.MLlib简述二.基本数据类型三.汇总统计四.实例 ...
## Spark学习之路（一）
前言: Spark 的核心是建立在统一的抽象弹性分布式数据集(Resiliennt Distributed Datasets,RDD)之上的,这使得 Spark 的各个组件可以无缝地进行集成,能够在同 ...
我的spark学习之路（三）：利用spark做回归分析
spark的机器学习库(MLlib)下有简单的回归分析方法,今天只说最简单的线性回归,spark提供有两个回归分析库(mllib和ml),我学习的时候在网上也查了不少资料,有一个奇怪的现象是网上关于s ...
Spark学习之路（五）Spark伪分布式安装
一.JDK的安装 JDK使用root用户安装 1.1 上传安装包并解压 [root@hadoop1 soft]# tar -zxvf jdk-8u73-linux-x64.tar.gz -C /usr ...
Java零基础学习之路（十三）Java super
Java super的定义本章节目标: 掌握super都可以用在哪里?理解super在内存方面的存储位置.掌握怎么通过子类的构造方法调用父类的构造方法.super什么时候可以省略,什么时候不能省略 ...

Spark学习之路(八)：分别使用Java与Scala实现Spark二次排序

内容简介

一、Spark二次排序的概念

二、实现二次排序的详细步骤(Java语言)

三、二次排序代码演示

1.Java版本

2.Scala版本

四、总结

Spark学习之路(八)：分别使用Java与Scala实现Spark二次排序相关推荐

最新文章

热门文章