案例四

编程那点事

发布于 2023-02-25 15:49:43

2650

发布于 2023-02-25 15:49:43

需求：对每个班级内的学生成绩，取出前3名（分组topn） class1 90 class2 56 class3 87 class1 76 class2 88 class1 95 class1 74 class2 87 class2 67 class2 77

Java版本

public class GroupTop3 {
public static void main(String[] args) {

SparkConf conf = new SparkConf().setAppName("Top3").setMaster("local");  
JavaSparkContext sc = new JavaSparkContext(conf);    
JavaRDD<String> lines = sc.textFile("C://Users//zhang//Desktop//score.txt");
JavaPairRDD<String, Integer> pairs = lines.mapToPair(
new PairFunction<String, String, Integer>() {

private static final long serialVersionUID = 1L;

 @Override
public Tuple2<String, Integer> call(String line) throws Exception {
String[] lineSplited = line.split(" ");  
return new Tuple2<String, Integer>(lineSplited[0],
Integer.valueOf(lineSplited[1]));
}
});

JavaPairRDD<String, Iterable<Integer>> groupedPairs = pairs.groupByKey();

JavaPairRDD<String, Iterable<Integer>> top3Score = groupedPairs.mapToPair(

new PairFunction<Tuple2<String,Iterable<Integer>>, String, Iterable<Integer>>() {

private static final long serialVersionUID = 1L;

@Override
public Tuple2<String, Iterable<Integer>> call(
Tuple2<String, Iterable<Integer>> classScores) throws Exception {
Integer[] top3 = new Integer[3];
String className = classScores._1;
Iterator<Integer> scores = classScores._2.iterator();
while(scores.hasNext()) {
Integer score = scores.next();
for(int i = 0; i < 3; i++) {
if(top3[i] == null) {
top3[i] = score;
break;
} else if(score > top3[i]) {
for(int j = 2; j > i; j--) {
top3[j] = top3[j - 1];  
}

top3[i] = score;
break;
}
}
}

return new Tuple2<String,
Iterable<Integer>>(className, Arrays.asList(top3));    
}
});
JavaPairRDD<String, Iterable<Integer>> sortedKeyTop3Score = top3Score.sortByKey();
sortedKeyTop3Score.foreach(new VoidFunction<Tuple2<String,Iterable<Integer>>>() {

private static final long serialVersionUID = 1L;

@Override
public void call(Tuple2<String, Iterable<Integer>> t) throws Exception {
System.out.println("class: " + t._1);  
Iterator<Integer> scoreIterator = t._2.iterator();
while(scoreIterator.hasNext()) {
Integer score = scoreIterator.next();
System.out.println(score);  
}
System.out.println("=======================================");  
}
});
sc.close();
}
 }

Scala版本

object GroupTop3 {

def main(args: Array[String]){

 val conf = new SparkConf().setAppName("Top3").setMaster("local")
  val sc = new SparkContext(conf)
 
  val lines = sc.textFile("C:/Users/zhang/Desktop/score.txt", 1)
 
  val pairs=lines.map { line => (line.split(" ")(0),line.split(" ")(1).toInt) }
 val grouped=pairs.groupByKey
 
 val groupedTop5=grouped.map(grouped=>
 {
   (grouped._1,grouped._2.toList.sortWith(_>_).take(2))
 })
val groupedKeySorted=groupedTop5.sortByKey()

 groupedKeySorted.foreach(pair=>{
 println(pair._1+":")
 pair._2.foreach { println }
 })
}
}

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2019-02-21，如有侵权请联系 cloudcommunity@tencent.com 删除

java