Spark：如何使用 spark 获得前 5 名答案

【问题标题】：Spark: how to get top 5 rec using sparkSpark：如何使用 spark 获得前 5 名
【发布时间】：2016-10-06 05:16:29
【问题描述】：

我正在转换 Spark 中的 MR 作业，并停留在我需要收集数据的地方，这些数据不在 RDD 中，而是在 scala 树图中。用例是通过标志中的条形（字段 8）和条形（字段 9）的总和来查找前 5 个国家/地区。所以我从源中提取数据并将其保存在树形图中。

请提出解决方案或解决此问题的新方法。

示例输入：

UK,3,4,245,56,1,1,0,0,3,1,0,1,0,1,0,0,red,0,1,1,0,0,0,0,0,0,0,white,red
Uruguay,2,3,178,3,2,0,0,9,3,0,0,1,1,1,0,0,white,0,0,0,1,1,0,0,0,0,0,white,white
US-Virgin-Isles,1,4,0,0,1,1,0,0,6,1,1,1,1,1,0,0,white,0,0,0,0,0,0,0,1,1,1,white,white
USA,1,4,9363,231,1,1,0,13,3,1,0,1,0,1,0,0,white,0,0,0,1,50,0,0,0,0,0,blue,red

MR 映射/减少：

public static class StripeBarMapper extends Mapper<Object, Text, IntWritable, Text> {

    private TreeMap<IntWritable, Text> Top5 = new TreeMap<IntWritable,Text>();

    @Override
    public void map(Object key, Text value, Context context) throws IOException, InterruptedException {

        String [] flags = value.toString().split(",");

        Top5.put(new IntWritable(Integer.parseInt(flags[7]) + Integer.parseInt(flags[8])), new Text(flags[0]));

        if(Top5.size() > 5){
            Top5.remove(Top5.firstKey());
        }
    }

    @Override
    protected void cleanup(Context context)
            throws IOException, InterruptedException {
        for ( IntWritable count : Top5.keySet()){
            context.write(count, Top5.get(count));
        }
    }
}

public  static class StripeBarReducer extends Reducer<IntWritable, Text, Text, Text> {

    public void reduce(Text key, Iterable<Text> values, Context context) 
            throws IOException, InterruptedException {

        System.out.print("reducer received: " + key.toString() + " --> ");

        for(Text country: values)
        context.write(new Text(country), new Text(key));
    }
}

激发工作

    val inputFile = "Data\\country\\flag.data"

    val conf = new SparkConf().setAppName("Top 5 Countries by Sum of bars and strips in flag").setMaster("local")
    val sc = new SparkContext(conf)

    val txtFileLines = sc.textFile(inputFile).cache()

    var tm = TreeMap(1 -> "one")

    val Tops =  txtFileLines.map(_.split(","))
                            .map{ s => 
                                  if(tm.size > 5){
                                    tm -= tm.firstKey
                                    }
                                   tm += ((s(7).toInt + s(8).toInt) -> s(0))
                                    }
                           //.sortBy( x => x.toString(), ascending = false, 1).saveAsTextFile("output\\country\\byStripsBar")
                           //.reduce(tm.keys.foreach { x => ($x._1, $x._2) })

解决方案： 我想遵循与 mapreduce 作业相同的方法是不合适的。我使用 sortBy 解决了这个问题：

val inputFile = "Data\\country\\flag.data"

val conf = new SparkConf().setAppName("Top 5 Countries by Sum of bars and strips in flag").setMaster("local")

val sc = new SparkContext(conf)

val txtFileLines = sc.textFile(inputFile).cache()

val Strips =  txtFileLines.map(_.split(","))
                         .map(line => (line(0) + "," + (line(7).toInt + line(8).toInt)))
                         .sortBy(x => x.split(",")(1).trim().toInt, ascending=false)
                         .take(5)
                       //.saveAsTextFile("output\\country\\byStripsBar")

 Strips.foreach { line => println(line) }

【问题讨论】：

您的实际问题是什么？代码哪里有问题？
问题：在 Sprk 工作中（请参阅我尝试作为解决方案但没有给我想要的结果的注释行）问题：如何从 scala TreeMap（从 RDD 中提取）获取数据，并使 RDD 到运行减速机。所以Tops map step 填充tm TreeMap，现在我想使用tm 中的值来获得最终输出或传递tm 的值来为reduce step 创建新的RDD。
我的解决方案按排序顺序存储来自 RDD 的所有记录，因此前 5 个记录是前 5 个国家/地区，但我只想存储前 5 个记录 - 是否可以在 RDD 中转换集合[take(5)]？
听起来像你想要的 RDD.top "def top(num: Int)(implicit ord: Ordering[T]): Array[T] 根据定义返回此 RDD 中的前 k 个（最大）元素通过指定的隐式 Ordering[T] 并保持排序。”。您的问题非常不清楚 - 尽管标题与 TreeMap 无关，但似乎因为您的解决方案没有使用它。

标签： scala apache-spark mapreduce bigdata

【解决方案1】：

我使用sortBy 和take() 解决了这个问题：

val inputFile = "Data\\country\\flag.data"

val conf = new SparkConf().setAppName("Top 5 Countries by Sum of bars and strips in flag").setMaster("local")

val sc = new SparkContext(conf)

val txtFileLines = sc.textFile(inputFile).cache()

val Strips =  txtFileLines.map(_.split(","))
                         .map(line => (line(0) + "," + (line(7).toInt + line(8).toInt)))
                         .sortBy(x => x.split(",")(1).trim().toInt, ascending=false)
                         .take(5)

Strips.foreach { line => println(line) }

sc.parallelize(Strips, 1).saveAsTextFile("output\\country\\byStripsBar")

【讨论】：