top N彻底解秘

1
4
2
5
7
3
2
7
9
1
4
5

/*** Return an array that contains all of the elements in this RDD.*/
def collect(): Array[T] = withScope {val results = sc.runJob(this, (iter: Iterator[T]) => iter.toArray)Array.concat(results: _*)
}

/*** Take the first num elements of the RDD. It works by first scanning one partition, and use the* results from that partition to estimate the number of additional partitions needed to satisfy* the limit.** @note due to complications in the internal implementation, this method will raise* an exception if called on an RDD of `Nothing` or `Null`.*/
def take(num: Int): Array[T] = withScope {if (num == 0) {new Array[T](0)} else {val buf = new ArrayBuffer[T]val totalParts = this.partitions.lengthvar partsScanned = 0while (buf.size < num && partsScanned < totalParts) {// The number of partitions to try in this iteration. It is ok for this number to be// greater than totalParts because we actually cap it at totalParts in runJob.var numPartsToTry = 1if (partsScanned > 0) {// If we didn't find any rows after the previous iteration, quadruple and retry.// Otherwise, interpolate the number of partitions we need to try, but overestimate// it by 50%. We also cap the estimation in the end.if (buf.size == 0) {numPartsToTry = partsScanned * 4} else {// the left side of max is >=1 whenever partsScanned >= 2numPartsToTry = Math.max((1.5 * num * partsScanned / buf.size).toInt - partsScanned, 1)numPartsToTry = Math.min(numPartsToTry, partsScanned * 4)}}val left = num - buf.sizeval p = partsScanned until math.min(partsScanned + numPartsToTry, totalParts)val res = sc.runJob(this, (it: Iterator[T]) => it.take(left).toArray, p)res.foreach(buf ++= _.take(num - buf.size))partsScanned += numPartsToTry}buf.toArray}
}

package com.zhouls.spark.coresimport org.apache.spark.{SparkConf, SparkContext}/*** 基础Top N实战* Created by Administrator on 2016/10/9.*/
object TopNBasic {def main(args: Array[String]) {val conf = new SparkConf()conf.setAppName("Top N Basically!").setMaster("local")val sc = new SparkContext(conf)val lines = sc.textFile("D://SoftWare//spark-1.5.2-bin-hadoop2.6//basicTopN.txt")val pairs = lines.map(line =>(line.toInt,line)) //生成key-value键值对，方便sortByKey进行排序val sortedPairs = pairs.sortByKey(false) //降序排序val sortedData = sortedPairs.map(pair => pair._2)  //只要是改变每一行列的数据，一般都是用map操作。过滤出排序后的内容本身val top5 = sortedData.take(5)  //获取排名前5位的元素内容
    top5.foreach(println)}
}

/** Control our logLevel. This overrides any user-defined log settings.* @param logLevel The desired log level as a string.* Valid log levels include: ALL, DEBUG, ERROR, FATAL, INFO, OFF, TRACE, WARN*/
def setLogLevel(logLevel: String) {val validLevels = Seq("ALL", "DEBUG", "ERROR", "FATAL", "INFO", "OFF", "TRACE", "WARN")if (!validLevels.contains(logLevel)) {throw new IllegalArgumentException(s"Supplied level $logLevel did not match one of: ${validLevels.mkString(",")}")}Utils.setLogLevel(org.apache.log4j.Level.toLevel(logLevel))
}

Spark 100
Hadoop 65
Spark 99
Hadoop 61
Spark 195
Hadoop 60
Spark 98
Hadoop 69
Spark 91
Hadoop 64
Spark 89
Hadoop 98
Spark 88 
Hadoop 99
Spark 68
Hadoop 60
Spark 79
Hadoop 97
Spark 69
Hadoop 96

package com.zhouls.spark.SparkApps.cores;import java.util.Arrays;
import java.util.Iterator;
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.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;public class TopNGroup {
public static void main(String[] args) {
SparkConf conf = new SparkConf().setAppName("TopNGroup").setMaster("local");
JavaSparkContext sc = new JavaSparkContext(conf); //其底层实际上就是Scala的SparkContext

JavaRDD<String> lines = sc.textFile("D://SoftWare//spark-1.5.2-bin-hadoop2.6//groupTopN.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[] splitedLine =line.split(" ");
System.out.println(splitedLine[0]);
return new Tuple2<String,Integer>(splitedLine[0],Integer.valueOf(splitedLine[1]));
}
});JavaPairRDD<String, Iterable<Integer>> groupedPairs =pairs.groupByKey();JavaPairRDD<String, Iterable<Integer>> top5=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>> groupedData)
throws Exception {
// TODO Auto-generated method stub

Integer[] top5=new Integer[5];
String groupedKey= groupedData._1;
Iterator<Integer> groupedValue = groupedData._2.iterator();while(groupedValue.hasNext()){
Integer value = groupedValue.next();for (int i =0; i<5; i++){
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<String, Iterable<Integer>>(groupedKey,Arrays.asList(top5));
}
}) ;//打印分组后的Top N 
top5.foreach(new VoidFunction<Tuple2<String,Iterable<Integer>>>() {
@Override
public void call(Tuple2<String, Iterable<Integer>> topped) throws Exception {System.out.println("Group key :"+ topped._1);//获取Group key
Iterator<Integer> toppedValue = topped._2.iterator(); //获取Group Value
while (toppedValue.hasNext()){ //具体打印出每组的Top N
Integer value =toppedValue.next();
System.out.println(value); 
}
System.out.println("******************************************************88");
}
});}
}

Spark 100
Hadoop 62
Flink 77
Kafka 91
Hadoop 93
Spark 78
Hadoop 69
Spark 98
Hadoop 62
Spark 99
Hadoop 61
Spark 70
Hadoop 75
Spark 88
Hadoop 68
Spark 90
Hadoop 61

/*** Sort the RDD by key, so that each partition contains a sorted range of the elements. Calling* `collect` or `save` on the resulting RDD will return or output an ordered list of records* (in the `save` case, they will be written to multiple `part-X` files in the filesystem, in* order of the keys).*/
// TODO: this currently doesn't work on P other than Tuple2!
def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length): RDD[(K, V)] = self.withScope
{val part = new RangePartitioner(numPartitions, self, ascending)new ShuffledRDD[K, V, V](self, part).setKeyOrdering(if (ascending) ordering else ordering.reverse)
}

/** Licensed to the Apache Software Foundation (ASF) under one or more* contributor license agreements.  See the NOTICE file distributed with* this work for additional information regarding copyright ownership.* The ASF licenses this file to You under the Apache License, Version 2.0* (the "License"); you may not use this file except in compliance with* the License.  You may obtain a copy of the License at**    http://www.apache.org/licenses/LICENSE-2.0** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an "AS IS" BASIS,* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.* See the License for the specific language governing permissions and* limitations under the License.*/package org.apache.sparkimport java.io.{IOException, ObjectInputStream, ObjectOutputStream}import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
import scala.reflect.{ClassTag, classTag}
import scala.util.hashing.byteswap32import org.apache.spark.rdd.{PartitionPruningRDD, RDD}
import org.apache.spark.serializer.JavaSerializer
import org.apache.spark.util.{CollectionsUtils, Utils}
import org.apache.spark.util.random.{XORShiftRandom, SamplingUtils}/*** An object that defines how the elements in a key-value pair RDD are partitioned by key.* Maps each key to a partition ID, from 0 to `numPartitions - 1`.*/
abstract class Partitioner extends Serializable {def numPartitions: Intdef getPartition(key: Any): Int
}object Partitioner {/*** Choose a partitioner to use for a cogroup-like operation between a number of RDDs.** If any of the RDDs already has a partitioner, choose that one.** Otherwise, we use a default HashPartitioner. For the number of partitions, if* spark.default.parallelism is set, then we'll use the value from SparkContext* defaultParallelism, otherwise we'll use the max number of upstream partitions.** Unless spark.default.parallelism is set, the number of partitions will be the* same as the number of partitions in the largest upstream RDD, as this should* be least likely to cause out-of-memory errors.** We use two method parameters (rdd, others) to enforce callers passing at least 1 RDD.*/def defaultPartitioner(rdd: RDD[_], others: RDD[_]*): Partitioner = {val bySize = (Seq(rdd) ++ others).sortBy(_.partitions.size).reversefor (r <- bySize if r.partitioner.isDefined && r.partitioner.get.numPartitions > 0) {return r.partitioner.get}if (rdd.context.conf.contains("spark.default.parallelism")) {new HashPartitioner(rdd.context.defaultParallelism)} else {new HashPartitioner(bySize.head.partitions.size)}}
}/*** A [[org.apache.spark.Partitioner]] that implements hash-based partitioning using* Java's `Object.hashCode`.** Java arrays have hashCodes that are based on the arrays' identities rather than their contents,* so attempting to partition an RDD[Array[_]] or RDD[(Array[_], _)] using a HashPartitioner will* produce an unexpected or incorrect result.*/
class HashPartitioner(partitions: Int) extends Partitioner {require(partitions >= 0, s"Number of partitions ($partitions) cannot be negative.")def numPartitions: Int = partitionsdef getPartition(key: Any): Int = key match {case null => 0case _ => Utils.nonNegativeMod(key.hashCode, numPartitions)}override def equals(other: Any): Boolean = other match {case h: HashPartitioner =>h.numPartitions == numPartitionscase _ =>false}override def hashCode: Int = numPartitions
}/*** A [[org.apache.spark.Partitioner]] that partitions sortable records by range into roughly* equal ranges. The ranges are determined by sampling the content of the RDD passed in.** Note that the actual number of partitions created by the RangePartitioner might not be the same* as the `partitions` parameter, in the case where the number of sampled records is less than* the value of `partitions`.*/
class RangePartitioner[K : Ordering : ClassTag, V](@transient partitions: Int,@transient rdd: RDD[_ <: Product2[K, V]],private var ascending: Boolean = true)extends Partitioner {// We allow partitions = 0, which happens when sorting an empty RDD under the default settings.require(partitions >= 0, s"Number of partitions cannot be negative but found $partitions.")private var ordering = implicitly[Ordering[K]]// An array of upper bounds for the first (partitions - 1) partitionsprivate var rangeBounds: Array[K] = {if (partitions <= 1) {Array.empty} else {// This is the sample size we need to have roughly balanced output partitions, capped at 1M.val sampleSize = math.min(20.0 * partitions, 1e6)// Assume the input partitions are roughly balanced and over-sample a little bit.val sampleSizePerPartition = math.ceil(3.0 * sampleSize / rdd.partitions.size).toIntval (numItems, sketched) = RangePartitioner.sketch(rdd.map(_._1), sampleSizePerPartition)if (numItems == 0L) {Array.empty} else {// If a partition contains much more than the average number of items, we re-sample from it// to ensure that enough items are collected from that partition.val fraction = math.min(sampleSize / math.max(numItems, 1L), 1.0)val candidates = ArrayBuffer.empty[(K, Float)]val imbalancedPartitions = mutable.Set.empty[Int]sketched.foreach { case (idx, n, sample) =>if (fraction * n > sampleSizePerPartition) {imbalancedPartitions += idx} else {// The weight is 1 over the sampling probability.val weight = (n.toDouble / sample.size).toFloatfor (key <- sample) {candidates += ((key, weight))}}}if (imbalancedPartitions.nonEmpty) {// Re-sample imbalanced partitions with the desired sampling probability.val imbalanced = new PartitionPruningRDD(rdd.map(_._1), imbalancedPartitions.contains)val seed = byteswap32(-rdd.id - 1)val reSampled = imbalanced.sample(withReplacement = false, fraction, seed).collect()val weight = (1.0 / fraction).toFloatcandidates ++= reSampled.map(x => (x, weight))}RangePartitioner.determineBounds(candidates, partitions)}}}def numPartitions: Int = rangeBounds.length + 1private var binarySearch: ((Array[K], K) => Int) = CollectionsUtils.makeBinarySearch[K]def getPartition(key: Any): Int = {val k = key.asInstanceOf[K]var partition = 0if (rangeBounds.length <= 128) {// If we have less than 128 partitions naive searchwhile (partition < rangeBounds.length && ordering.gt(k, rangeBounds(partition))) {partition += 1}} else {// Determine which binary search method to use only once.partition = binarySearch(rangeBounds, k)// binarySearch either returns the match location or -[insertion point]-1if (partition < 0) {partition = -partition-1}if (partition > rangeBounds.length) {partition = rangeBounds.length}}if (ascending) {partition} else {rangeBounds.length - partition}}override def equals(other: Any): Boolean = other match {case r: RangePartitioner[_, _] =>r.rangeBounds.sameElements(rangeBounds) && r.ascending == ascendingcase _ =>false}override def hashCode(): Int = {val prime = 31var result = 1var i = 0while (i < rangeBounds.length) {result = prime * result + rangeBounds(i).hashCodei += 1}result = prime * result + ascending.hashCoderesult}@throws(classOf[IOException])private def writeObject(out: ObjectOutputStream): Unit = Utils.tryOrIOException {val sfactory = SparkEnv.get.serializersfactory match {case js: JavaSerializer => out.defaultWriteObject()case _ =>out.writeBoolean(ascending)out.writeObject(ordering)out.writeObject(binarySearch)val ser = sfactory.newInstance()Utils.serializeViaNestedStream(out, ser) { stream =>stream.writeObject(scala.reflect.classTag[Array[K]])stream.writeObject(rangeBounds)}}}@throws(classOf[IOException])private def readObject(in: ObjectInputStream): Unit = Utils.tryOrIOException {val sfactory = SparkEnv.get.serializersfactory match {case js: JavaSerializer => in.defaultReadObject()case _ =>ascending = in.readBoolean()ordering = in.readObject().asInstanceOf[Ordering[K]]binarySearch = in.readObject().asInstanceOf[(Array[K], K) => Int]val ser = sfactory.newInstance()Utils.deserializeViaNestedStream(in, ser) { ds =>implicit val classTag = ds.readObject[ClassTag[Array[K]]]()rangeBounds = ds.readObject[Array[K]]()}}}
}private[spark] object RangePartitioner {/*** Sketches the input RDD via reservoir sampling on each partition.** @param rdd the input RDD to sketch* @param sampleSizePerPartition max sample size per partition* @return (total number of items, an array of (partitionId, number of items, sample))*/def sketch[K : ClassTag](rdd: RDD[K],sampleSizePerPartition: Int): (Long, Array[(Int, Int, Array[K])]) = {val shift = rdd.id// val classTagK = classTag[K] // to avoid serializing the entire partitioner objectval sketched = rdd.mapPartitionsWithIndex { (idx, iter) =>val seed = byteswap32(idx ^ (shift << 16))val (sample, n) = SamplingUtils.reservoirSampleAndCount(iter, sampleSizePerPartition, seed)Iterator((idx, n, sample))}.collect()val numItems = sketched.map(_._2.toLong).sum(numItems, sketched)}/*** Determines the bounds for range partitioning from candidates with weights indicating how many* items each represents. Usually this is 1 over the probability used to sample this candidate.** @param candidates unordered candidates with weights* @param partitions number of partitions* @return selected bounds*/def determineBounds[K : Ordering : ClassTag](candidates: ArrayBuffer[(K, Float)],partitions: Int): Array[K] = {val ordering = implicitly[Ordering[K]]val ordered = candidates.sortBy(_._1)val numCandidates = ordered.sizeval sumWeights = ordered.map(_._2.toDouble).sumval step = sumWeights / partitionsvar cumWeight = 0.0var target = stepval bounds = ArrayBuffer.empty[K]var i = 0var j = 0var previousBound = Option.empty[K]while ((i < numCandidates) && (j < partitions - 1)) {val (key, weight) = ordered(i)cumWeight += weightif (cumWeight > target) {// Skip duplicate values.if (previousBound.isEmpty || ordering.gt(key, previousBound.get)) {bounds += keytarget += stepj += 1previousBound = Some(key)}}i += 1}bounds.toArray}
}

/*** Sketches the input RDD via reservoir sampling on each partition.** @param rdd the input RDD to sketch* @param sampleSizePerPartition max sample size per partition* @return (total number of items, an array of (partitionId, number of items, sample))*/
def sketch[K : ClassTag](rdd: RDD[K],sampleSizePerPartition: Int): (Long, Array[(Int, Int, Array[K])]) = {val shift = rdd.id// val classTagK = classTag[K] // to avoid serializing the entire partitioner objectval sketched = rdd.mapPartitionsWithIndex { (idx, iter) =>val seed = byteswap32(idx ^ (shift << 16))val (sample, n) = SamplingUtils.reservoirSampleAndCount(iter, sampleSizePerPartition, seed)Iterator((idx, n, sample))}.collect()val numItems = sketched.map(_._2.toLong).sum(numItems, sketched)
}

/*** Reservoir sampling implementation that also returns the input size.** @param input input size* @param k reservoir size* @param seed random seed* @return (samples, input size)*/
def reservoirSampleAndCount[T: ClassTag](input: Iterator[T],k: Int,seed: Long = Random.nextLong()): (Array[T], Int) = {val reservoir = new Array[T](k)// Put the first k elements in the reservoir.var i = 0while (i < k && input.hasNext) {val item = input.next()reservoir(i) = itemi += 1}// If we have consumed all the elements, return them. Otherwise do the replacement.if (i < k) {// If input size < k, trim the array to return only an array of input size.val trimReservoir = new Array[T](i)System.arraycopy(reservoir, 0, trimReservoir, 0, i)(trimReservoir, i)} else {// If input size > k, continue the sampling process.val rand = new XORShiftRandom(seed)while (input.hasNext) {val item = input.next()val replacementIndex = rand.nextInt(i)if (replacementIndex < k) {reservoir(replacementIndex) = item}i += 1}(reservoir, i)}
}

def getPartition(key: Any): Int = {val k = key.asInstanceOf[K]var partition = 0if (rangeBounds.length <= 128) {// If we have less than 128 partitions naive searchwhile (partition < rangeBounds.length && ordering.gt(k, rangeBounds(partition))) {partition += 1}} else {// Determine which binary search method to use only once.partition = binarySearch(rangeBounds, k)// binarySearch either returns the match location or -[insertion point]-1if (partition < 0) {partition = -partition-1}if (partition > rangeBounds.length) {partition = rangeBounds.length}}if (ascending) {partition} else {rangeBounds.length - partition}
}