溫馨提示×
您好，登錄后才能下訂單哦！
點(diǎn)擊 登錄注冊 即表示同意《億速云用戶服務(wù)條款》
您好，登錄后才能下訂單哦！
本節(jié)課內(nèi)容：
1、基礎(chǔ)排序算法實(shí)戰(zhàn)
2、二次排序算法實(shí)戰(zhàn)
3、更高級別排序算法
4、排序算法內(nèi)幕解密
排序在Spark運(yùn)用程序中使用的比較多，且維度也不一樣，如二次排序，三次排序等，在機(jī)器學(xué)習(xí)算法中經(jīng)常碰到，所以非常重要，必須掌握！
所謂二次排序，就是根據(jù)兩列值進(jìn)行排序，如下測試數(shù)據(jù)：
2 3
4 1
3 2
4 3
8 7
2 1
經(jīng)過二次排序后的結(jié)果（升序）：
2 1
2 3
3 2
4 1
4 3
8 7
在編寫二次排序代碼前，先簡單的寫下單個(gè)key排序的代碼：
val conf = new SparkConf().setAppName("SortByKey").setMaster("local")
val sc = new SparkContext(conf)
val lines = sc.textFile("C:\\User\\Test.txt")
valwords = lines.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)val wordcount = words.map(word=>(word._2,word._1)).sortByKey(false).map(word=>(word._2,word._1))
wordcount.collect().foreach(println)
以上就是簡單的wordcount程序，程序中使用了sortByKey排序
下面我們通過代碼實(shí)現(xiàn)二次排序算法
首先我們先通過Java代碼實(shí)現(xiàn)上面測試數(shù)據(jù)進(jìn)行二次排序
排序最主要的就是Key的準(zhǔn)備，我們先用Java編寫二次排序的key，參考代碼如下：
import java.io.Serializable;
import scala.math.Ordered;
public class SecondarySortKey implements Ordered<SecondarySortKey>, Serializable {
private int first;
private int second;
@Override
public int hashCode() {
final int prime = 31;
int result = 1;
result = prime * result + first;
result = prime * result + second;
return result;
}
@Override
public boolean equals(Object obj) {
if (this == obj)
return true;
if (obj == null)
return false;
if (getClass() != obj.getClass())
return false;
SecondarySortKey other = (SecondarySortKey) obj;
if (first != other.first)
return false;
if (second != other.second)
return false;
return true;
}
public int getFirst() {
return first;
}
public void setFirst(int first) {
this.first = first;
}
public int getSecond() {
return second;
}
public void setSecond(int second) {
this.second = second;
}
public SecondarySortKey(int first, int second) {
this.first = first;
this.second = second;
}
public boolean $greater(SecondarySortKey other) {
if (this.first > other.getFirst()) {
return true;
} else if (this.first == other.getFirst() && this.second > other.getSecond()) {
return true;
}
return false;
}
public boolean $greater$eq(SecondarySortKey other) {
if (this.$greater(other)) {
return true;
} else if (this.first == other.getFirst() && this.second == other.getSecond()) {
return true;
}
return false;
}
public boolean $less(SecondarySortKey other) {
if (this.first < other.getFirst()) {
return true;
} else if (this.first == other.getFirst() && this.second < other.getSecond()) {
return true;
}
return false;
}
public boolean $less$eq(SecondarySortKey other) {
if (this.$less(other)) {
return true;
} else if (this.first == other.getFirst() && this.second < other.getSecond()) {
return true;
}
return false;
}
public int compare(SecondarySortKey other) {
if (this.first - other.getFirst() != 0) {
return this.first - other.getFirst();
} else {
return this.second - other.getSecond();
}
}
public int compareTo(SecondarySortKey other) {
if (this.first - other.getFirst() != 0) {
return this.first - other.getFirst();
} else {
return this.second - other.getSecond();
}
}
根據(jù)上面生成的排序key編寫對測試數(shù)據(jù)的二次排序
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;
/**
* DT_Spark大數(shù)據(jù)夢工廠
* 二次排序，具體的實(shí)現(xiàn)步驟：
* 第一步：按照Ordered和Serializable接口實(shí)現(xiàn)自定義排序的key
* 第二步：將要進(jìn)行二次排序的文件加載進(jìn)來生成<key,value>類型的RDD
* 第三步：使用sortByKey基于自定義的Key進(jìn)行二次排序
* 第四步：去除掉排序的Key,只保留排序的結(jié)果
*/
public class SecondarySortKeyApp {
public static void main(String[] args) {
SparkConf conf = new SparkConf().setAppName("SecondarySortKeyApp").setMaster("local");
JavaSparkContext sc = new JavaSparkContext(conf);
JavaRDD<String> lines = sc.textFile("C:\\Users\\Test.txt");
//將自定義的key添加進(jìn)來
JavaPairRDD<SecondarySortKey, String> pairs = lines
.mapToPair(new PairFunction<String, SecondarySortKey, String>() {
private static final long serialVersionUID = 1L;
public Tuple2<SecondarySortKey, String> call(String line) throws Exception {
String[] splited = line.split(" ");
SecondarySortKey key = new SecondarySortKey(Integer.valueOf(splited[0]),
Integer.valueOf(splited[1]));
return new Tuple2<SecondarySortKey, String>(key, line);
}
});
//根據(jù)我們自定義的key進(jìn)行升序排序
JavaPairRDD<SecondarySortKey, String> sorted = pairs.sortByKey(); //sortByKey(false) 降序
// 過濾掉排序后的自定義的Key,保留排序的結(jié)果
JavaRDD<String> secondarySort = sorted.map(new Function<Tuple2<SecondarySortKey, String>, String>() {
public String call(Tuple2<SecondarySortKey, String> sortedContent) throws Exception {
return sortedContent._2;
}
});
secondarySort.foreach(new VoidFunction<String>() {
public void call(String sorted) throws Exception {
System.out.println(sorted);
}
});
}
運(yùn)行結(jié)果：
2 1
2 3
3 2
4 1
4 3
8 7
下面我通過Scala方式實(shí)現(xiàn)上述二次排序，scala代碼非常簡潔
先創(chuàng)建我們自定義排序key
*DT_Spark大數(shù)據(jù)夢工廠
* 自定義二次排序的key
*/
class SecondarySortKey(val first: Int, val second: Int) extends Ordered[SecondarySortKey] with Serializable {
def compare(other: SecondarySortKey): Int = {
if (this.first - other.first != 0) {
this.first - other.first
}
else {
this.second - other.second
}
}
根據(jù)自定義排序Key實(shí)現(xiàn)二次排序
import org.apache.spark.{SparkConf, SparkContext}
/**SecondarySortKeyApp {
def main(args: Array[String]) {//過濾掉key,只保留value
val sortedResult = sorted.map(sort => sort._2)
//顯示結(jié)果
sortedResult.collect().foreach(println)
}
}
運(yùn)行結(jié)果：
2 1
2 3
3 2
4 1
4 3
8 7
從上面的代碼可以看出，通過scala代碼實(shí)現(xiàn)二次排序確實(shí)非常簡潔，這也是scala的強(qiáng)大之處所在。
更高級別排序算法和內(nèi)幕解密在后續(xù)課程中在分享。
備注：
資料來源于：DT_大數(shù)據(jù)夢工廠
更多私密內(nèi)容，請關(guān)注微信公眾號(hào)：DT_Spark
如果您對大數(shù)據(jù)Spark感興趣，可以免費(fèi)聽由王家林老師每天晚上20：00開設(shè)的Spark永久免費(fèi)公開課，地址YY房間號(hào)：68917580
免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀點(diǎn)不代表本網(wǎng)站立場，如果涉及侵權(quán)請聯(lián)系站長郵箱：is@yisu.com進(jìn)行舉報(bào)，并提供相關(guān)證據(jù)，一經(jīng)查實(shí)，將立刻刪除涉嫌侵權(quán)內(nèi)容。
億速云公眾號(hào)
手機(jī)網(wǎng)站二維碼
