Spark Top N

基本TopN(Scala版)

package com.dt.spark.cores.scala

import org.apache.spark.{SparkContext, SparkConf}


object TopNBasic {
def main (args: Array[String]) {
val conf = new SparkConf()
conf.setAppName("TopNBasic")
conf.setMaster("local")
val sc = new SparkContext(conf)
val data = sc.textFile("E:\\workspases\\data\\basicTopN.txt")

val pairs = data.map(line => (line.toInt,line))

val sortedPairs = pairs.sortByKey(false)

val sortedData = sortedPairs.map(pair => pair._2)

val top5 = sortedData.take(5)

sc.setLogLevel("OFF")
top5.foreach(println)

sc.stop()
}
}

topNグループ(Java版):

package com.dt.spark.cores.java;

import com.sun.corba.se.spi.legacy.connection.GetEndPointInfoAgainException;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
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.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;


public class TopNGroup {
public static void main(String[] args)
{
SparkConf conf = new SparkConf().setAppName("TopNGroup").setMaster("local");

JavaSparkContext sc = new JavaSparkContext(conf);

JavaRDD data = sc.textFile("E:\\workspases\\data\\GroupTopN.txt");

JavaPairRDD,Integer> pairs = data.mapToPair(new PairFunction, String, Integer>() {
@Override
public Tuple2, Integer> call(String line) throws Exception {
String[] splitedline = line.split(" ");
return new Tuple2, Integer>(splitedline[0], Integer.valueOf(splitedline[1]));
}
});

JavaPairRDD,Iterable> groupPairs = pairs.groupByKey();

JavaPairRDD,Iterable> top5 =groupPairs.mapToPair(new PairFunction, Iterable>, String, Iterable>() {
@Override
public Tuple2, Iterable> call(Tuple2, Iterable> groupedData) throws Exception {
Integer[] top5 = new Integer[5];
String groupedKey = groupedData._1();
Iterator groupValue = groupedData._2().iterator();
while (groupValue.hasNext()) {
Integer value = groupValue.next();
for (int i = 0; i < 5; i++) {//       topN
if (top5[i] == null) {
top5[i] = value;
break;
} else if (value > top5[i]) {
for (int j = 4; j > i; j--) {
top5[j] = top5[j - 1];
}
top5[i] = value;
break;
}
}
}
return new Tuple2, Iterable>(groupedKey, Arrays.asList(top5));
}
});
sc.setLogLevel("OFF");
//      
top5.foreach(new VoidFunction, Iterable>>() {
@Override
public void call(Tuple2, Iterable> topped) throws Exception {
System.out.println("groupKey:" + topped._1());//  groupKey
Iterator toppedValue = topped._2().iterator();//  groupvalue
while (toppedValue.hasNext())//       Top N
{
Integer value = toppedValue.next();
System.out.println(value);
}
System.out.println("***********************************************");
}
});
}
}

RangePartitionerは主に依存するRDDのデータが異なる範囲に分けられていますが、肝心なところは異なる範囲が整然としているRangePartitionerは結果が整然とした礎である以外は、できるだけ各Partitionのデータ量が均一であることを保証することが重要です!!!HashPartitionではデータの傾きが発生し、極端にはある(いくつかの)パーティションがRDDのすべてのデータを持っている!!!ジョブ:Scalaを使用してTopNパケットプログラムを書き、Keyをソートする

package com.dt.spark.cores.scala

import org.apache.spark.{SparkContext, SparkConf}


object TopNGroup {
def main (args: Array[String]) {

val conf = new SparkConf()
conf.setAppName("TopNBasic")
conf.setMaster("local")
val sc = new SparkContext(conf)
val data = sc.textFile("E:\\workspases\\data\\GroupTopN.txt")

val groupRDD = data.map(line => (line.split(" ")(0),line.split(" ")(1).toInt)).groupByKey()

val top5 = groupRDD.map(pair=> (pair._1,pair._2.toList.sortWith(_>_).take(5))).sortByKey()
top5.collect().foreach(pair =>{
println(pair._1+":")
pair._2.foreach(println)
println("***********************")
})
sc.stop() 
}
}