Spark高级排序与TopN问题揭密

[TOC]

引入

前面进行过wordcount的单词统计例子，关键是，如何对统计的单词按照单词个数来进行排序？

如下：

scala> val retRDD = sc.textFile("hdfs://ns1/hello").flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_)

scala> val retSortRDD = retRDD.map(pair => (pair._2, pair._1)).sortByKey(false).map(pair => (pair._2, pair._1))

scala> retSortRDD.collect().foreach(println)
...
(hello,3)
(me,1)
(you,1)
(he,1)

下面的测试都需要引入maven的依赖

<dependency>
    <groupId>org.scala-lang</groupId>
    <artifactId>scala-library</artifactId>
    <version>2.10.5</version>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.10</artifactId>
    <version>1.6.2</version>
</dependency>

Spark二次排序

测试数据与说明

需要进行二次排序的数据格式如下：

field_1‘ ‘field_2(使用空格分割)
 20 21
 50 51
 50 52
 50 53
 50 54
 60 51
 60 53
 60 52
 60 56
 60 57
 70 58
 60 61
 70 54

思路下面的代码注释会有详细的说明，这里要指出的是，在下面的排序过程中，分别使用Java和Scala进行排序的操作，并且：

Java版本
- 方式1：使元素具备比较性--->需要使用SecondarySort对象
- 方式2：提供比较器--->需要使用SecondarySort对象
- 不管使用哪一种方式，都需要使用一个新的变量对象SecondarySort
Scala版本
- 方式1：使元素具备比较性，其实就是Java版本方式1的scala实现--->需要使用SecondarySort对象
- 方式2：使用sortBy的第一种方式，基于原始的数据进行排序--->不需要使用SecondarySort对象
- 方式3：使用sortBy的第二种方式，将原始数据进行转换--->需要使用SecondarySort对象

所以这个二次排序的例子包含Java和Scala总共5个版本的实现，非常有价值！

公共对象

其实就是SecondarySort对象，如下：

package cn.xpleaf.bigdata.spark.java.core.domain;

import scala.Serializable;

public class SecondarySort implements Comparable<SecondarySort>, Serializable {
    private int first;
    private int second;

    public SecondarySort(int first, int second) {
        this.first = first;
        this.second = second;
    }

    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;
    }

    @Override
    public int compareTo(SecondarySort that) {
        int ret = this.getFirst()  - that.getFirst();
        if(ret == 0) {
            ret = that.getSecond() - this.getSecond();
        }
        return ret;
    }

    @Override
    public String toString() {
        return this.first + " " + this.second;
    }
}

Java版本

测试代码如下：

package cn.xpleaf.bigdata.spark.java.core.p3;

import cn.xpleaf.bigdata.spark.java.core.domain.SecondarySort;
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.Serializable;
import scala.Tuple2;

import java.util.Comparator;

/**
 * Java 版本的二次排序
 *   field_1‘ ‘field_2(使用空格分割)
 *   20 21
 50 51
 50 52
 50 53
 50 54
 60 51
 60 53
 60 52
 60 56
 60 57
 70 58
 60 61
 70 54
 需求：首先按照第一列升序排序，如果第一列相等，按照第二列降序排序
 分析：要排序的话，使用sortByKey，也可以使用sortBy
 如果用sortByKey的话，只能按照key来排序，现在的是用第一列做key？还是第二列？
 根据需求，只能使用复合key（既包含第一列，也包含第二列），因为要进行比较，所以该复合key必须具备比较性，要么该操作提供一个比较器
 问题是查看该操作的时候，并没有给我们提供比较器，没得选只能让元素具备比较性

 使用自定义的对象 可以使用comprable接口
 */
public class _01SparkSecondarySortOps {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setMaster("local[2]").setAppName(_01SparkSecondarySortOps.class.getSimpleName());
        JavaSparkContext jsc = new JavaSparkContext(conf);

        JavaRDD<String> linesRDD = jsc.textFile("D:/data/spark/secondsort.csv");
        JavaPairRDD<SecondarySort, String> ssRDD = linesRDD.mapToPair(new PairFunction<String, SecondarySort, String>() {
            @Override
            public Tuple2<SecondarySort, String> call(String line) throws Exception {
                String[] fields = line.split(" ");
                int first = Integer.valueOf(fields[0].trim());
                int second = Integer.valueOf(fields[1].trim());
                SecondarySort ss = new SecondarySort(first, second);
                return new Tuple2<SecondarySort, String>(ss, "");
            }
        });
        /*
        // 第一种方式：使元素具备比较性
        JavaPairRDD<SecondarySort, String> sbkRDD = ssRDD.sortByKey(true, 1);   // 设置partition为1，这样数据才整体有序，否则只是partition中有序
        */

        /**
         * 第二种方式，提供比较器
         *      与前面方式相反，这次是：第一列降序，第二列升序
         */
        JavaPairRDD<SecondarySort, String> sbkRDD = ssRDD.sortByKey(new MyComparator<SecondarySort>() {
            @Override
            public int compare(SecondarySort o1, SecondarySort o2) {
                int ret = o2.getFirst() - o1.getFirst();
                if(ret == 0) {
                    ret = o1.getSecond() - o2.getSecond();
                }
                return ret;
            }
        }, true, 1);

        sbkRDD.foreach(new VoidFunction<Tuple2<SecondarySort, String>>() {
            @Override
            public void call(Tuple2<SecondarySort, String> tuple2) throws Exception {
                System.out.println(tuple2._1);
            }
        });

        jsc.close();
    }
}

/**
 * 做一个中间的过渡接口
 * 比较需要实现序列化接口，否则也会报异常
 * 是用到了适配器Adapter模式
 * 适配器模式(Adapter Pattern)是作为两个不兼容的接口之间的桥梁，这里就是非常好的体现了。
 */
interface MyComparator<T> extends Comparator<T>, Serializable{}

输出结果如下：

Scala版本

测试代码如下：

package cn.xpleaf.bigdata.spark.scala.core.p3

import cn.xpleaf.bigdata.spark.java.core.domain.SecondarySort
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.reflect.ClassTag

object _05SparkSecondarySortOps {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local[2]").setAppName(_05SparkSecondarySortOps.getClass.getSimpleName)
        val sc = new SparkContext(conf)

        val linesRDD = sc.textFile("D:/data/spark/secondsort.csv")

        /*
        val ssRDD:RDD[(SecondarySort, String)] = linesRDD.map(line => {
            val fields = line.split(" ")
            val first = Integer.valueOf(fields(0).trim())
            val second = Integer.valueOf(fields(1).trim())
            val ss = new SecondarySort(first, second)
            (ss, "")
        })

        // 第一种方式，使用元素具备比较性
        val sbkRDD:RDD[(SecondarySort, String)] = ssRDD.sortByKey(true, 1)

        sbkRDD.foreach{case (ss:SecondarySort, str:String) => { // 使用模式匹配的方式
            println(ss)
        }}
        */

        /*
        // 使用sortBy的第一种方式，基于原始的数据
        val retRDD = linesRDD.sortBy(line => line, numPartitions = 1)(new Ordering[String] {
            override def compare(x: String, y: String): Int = {
                val xFields = x.split(" ")
                val yFields = y.split(" ")

                var ret = xFields(0).toInt - yFields(0).toInt
                if(ret == 0) {
                    ret = yFields(1).toInt - xFields(1).toInt
                }
                ret
            }
        }, ClassTag.Object.asInstanceOf[ClassTag[String]])
        */

        // 使用sortBy的第二种方式，将原始数据做转换--->sortBy()第一个参数的作用，就是做数据的转换
        val retRDD:RDD[String] = linesRDD.sortBy(line => {
            // f: (T) => K
            // 这里T的类型为String，K是SecondarySort类型
            val fields = line.split(" ")
            val first = Integer.valueOf(fields(0).trim())
            val second = Integer.valueOf(fields(1).trim())
            val ss = new SecondarySort(first, second)
            ss
        }, true, 1)(new Ordering[SecondarySort] {
            override def compare(x: SecondarySort, y: SecondarySort): Int = {
                var ret = x.getFirst - y.getFirst
                if(ret == 0) {
                    ret = y.getSecond - x.getSecond
                }
                ret
            }
        }, ClassTag.Object.asInstanceOf[ClassTag[SecondarySort]])

        retRDD.foreach(println)

        sc.stop()
    }
}

输出结果如下：

TopN问题

需求与说明

需求与数据说明如下：

  * TopN问题的说明：
  *     TopN问题显然是可以使用action算子take来完成，但是因为take需要将所有数据都拉取到Driver上才能完成操作，
  *     所以Driver的内存压力非常大，不建议使用take.
  *
  * 这里要进行TopN问题的分析，数据及需求如下：
  * chinese ls 91
  * english ww 56
  * chinese zs 90
  * chinese zl 76
  * english zq 88
  * chinese wb 95
  * chinese sj 74
  * english ts 87
  * english ys 67
  * english mz 77
  * chinese yj 98
  * english gk 96
  *
  * 需求：排出每个科目的前三名

下面分别使用性能很低的groupByKey和性能很好的combineByKey来进行操作，详细的说明已经在代码中给出，注意其思想非常重要，尤其是使用combineByKey来解决groupByKey出现的性能问题，有兴趣的话，可以好好阅读一下代码，以及其所体现的思想，因为这都跟Spark本身的理论紧密相关。

使用groupByKey解决

测试代码如下：

package cn.xpleaf.bigdata.spark.scala.core.p3

import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

/**
  * TopN问题的说明：
  *     TopN问题显然是可以使用action算子take来完成，但是因为take需要将所有数据都拉取到Driver上才能完成操作，
  *     所以Driver的内存压力非常大，不建议使用take.
  *
  * 这里要进行TopN问题的分析，数据及需求如下：
  * chinese ls 91
  * english ww 56
  * chinese zs 90
  * chinese zl 76
  * english zq 88
  * chinese wb 95
  * chinese sj 74
  * english ts 87
  * english ys 67
  * english mz 77
  * chinese yj 98
  * english gk 96
  *
  * 需求：排出每个科目的前三名
  *
  * 思路：先进行map操作转换为(subject, name + score)的元组
  * 再根据subject这个key进行groupByKey，这样就可以得到gbkRDD
  * 之后再对其进行map操作，在map操作中使用treeSet得到前三名（既能控制大小，又能进行排序）
  *
  * 问题：
  * 上面的方案在生产过程中慎用
  * 因为，执行groupByKey，会将key相同的数据都拉取到同一个partition中，再执行操作，
  * 拉取的过程是shuffle，是分布式性能杀手！再一个，如果key对应的数据过多，很有可能造成数据倾斜，或者OOM，
  * 那么就需要尽量的避免这种操作方式。
  * 那如何做到？可以参考MR中TopN问题的思想，MR中，是在每个map task中对数据进行筛选，虽然最后还是需要shuffle到一个节点上，但是数据量会大大减少。
  * Spark中参考其中的思想，就是可以在每个partition中对数据进行筛选，然后再对各个分区筛选出来的数据进行合并，再做一次排序，从而得到最终排序的结果。
  * 显然，这样就可以解决前面说的数据到同一个partition中导致数据量过大的问题！因为分区筛选的工作已经可以大大减少数据量。
  * 那么在Spark中有什么算子可以做到这一点呢？那就是combineByKey或者aggregateByKey，其具体的用法可以参考我前面的博客文章，这里我使用combineByKey来操作。
  */
object _06SparkTopNOps {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local[2]").setAppName(_06SparkTopNOps.getClass.getSimpleName())
        val sc = new SparkContext(conf)
        Logger.getLogger("org.apache.spark").setLevel(Level.OFF)
        Logger.getLogger("org.apache.hadoop").setLevel(Level.OFF)
        Logger.getLogger("org.spark_project").setLevel(Level.OFF)

        // 1.转换为linesRDD
        val linesRDD:RDD[String] = sc.textFile("D:/data/spark/topn.txt")

        // 2.转换为pairsRDD
        val pairsRDD:RDD[(String, String)] = linesRDD.map(line => {
            val fields = line.split(" ")
            val subject = fields(0).trim()
            val name = fields(1).trim()
            val score = fields(2).trim()
            (subject, name + " " + score)   // ("chinese", "zs 90")
        })

        // 3.转换为gbkRDD
        val gbkRDD:RDD[(String, Iterable[String])] = pairsRDD.groupByKey()
        println("==========TopN前==========")
        gbkRDD.foreach(println)
        // (english,CompactBuffer(ww 56, zq 88, ts 87, ys 67, mz 77, gk 96))
        // (chinese,CompactBuffer(ls 91, zs 90, zl 76, wb 95, sj 74, yj 98))

        // 4.转换为retRDD
        val retRDD:RDD[(String, Iterable[String])] = gbkRDD.map(tuple => {
            var ts = new mutable.TreeSet[String]()(new MyOrdering())
            val subject = tuple._1          // chinese
            val nameScores = tuple._2       //  ("ls 91", "ww 56", "zs 90", ...)
            for(nameScore <- nameScores) {  // 遍历每一份成绩"ls 91"
                // 添加到treeSet中
                ts.add(nameScore)
                if(ts.size > 3) {   // 如果大小大于3，则弹出最后一份成绩
                    ts = ts.dropRight(1)
                }
            }
            (subject, ts)
        })

        println("==========TopN后==========")
        retRDD.foreach(println)

        sc.stop()
    }
}

// gbkRDD.map中用于排序的treeSet的排序比较规则，根据需求，应该为降序
class MyOrdering extends Ordering[String] {
    override def compare(x: String, y: String): Int = {
        // x或者y的格式为："zs 90"
        val xFields = x.split(" ")
        val yFields = y.split(" ")
        val xScore = xFields(1).toInt
        val yScore = yFields(1).toInt
        val ret = yScore - xScore
        ret
    }
}

输出结果如下：

==========TopN前==========
(chinese,CompactBuffer(ls 91, zs 90, zl 76, wb 95, sj 74, yj 98))
(english,CompactBuffer(ww 56, zq 88, ts 87, ys 67, mz 77, gk 96))
==========TopN后==========
(chinese,TreeSet(yj 98, wb 95, ls 91))
(english,TreeSet(gk 96, zq 88, ts 87))

使用combineByKey解决

测试代码如下：

package cn.xpleaf.bigdata.spark.scala.core.p3

import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

import scala.collection.mutable

/**
  * 使用combineByKey算子来优化前面的TopN问题
  * 关于combineByKey算子的使用，可以参考我的博客文章，上面有非常详细的例子
  * 一定要掌握，因为非常重要
  */
object _07SparkTopNOps {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local[2]").setAppName(_07SparkTopNOps.getClass().getSimpleName())
        val sc = new SparkContext(conf)
        Logger.getLogger("org.apache.spark").setLevel(Level.OFF)
        Logger.getLogger("org.apache.hadoop").setLevel(Level.OFF)
        Logger.getLogger("org.spark_project").setLevel(Level.OFF)

        // 1.转换为linesRDD
        val linesRDD:RDD[String] = sc.textFile("D:/data/spark/topn.txt")

        // 2.转换为pairsRDD
        val pairsRDD:RDD[(String, String)] = linesRDD.map(line => {
            val fields = line.split(" ")
            val subject = fields(0).trim()
            val name = fields(1).trim()
            val score = fields(2).trim()
            (subject, name + " " + score)   // ("chinese", "zs 90")
        })

        println("==========TopN前==========")
        pairsRDD.foreach(println)
        // (chinese,sj 74)
        // (chinese,ls 91)
        // (english,ts 87)
        // (english,ww 56)
        // (english,ys 67)
        // (chinese,zs 90)
        // (english,mz 77)
        // (chinese,zl 76)
        // (chinese,yj 98)
        // (english,zq 88)
        // (english,gk 96)
        // (chinese,wb 95)

        // 3.转换为cbkRDD
        val cbkRDD:RDD[(String, mutable.TreeSet[String])] = pairsRDD.combineByKey(createCombiner, mergeValue, mergeCombiners)

        println("==========TopN后==========")
        cbkRDD.foreach(println)
        // (chinese,TreeSet(yj 98, wb 95, ls 91))
        // (english,TreeSet(gk 96, zq 88, ts 87))
    }

    // 创建一个容器，这里返回一个treeSet，作为每个分区中相同key的value的容器
    def createCombiner(nameScore: String):mutable.TreeSet[String] = {
        // nameScore格式为："zs 90"
        // 指定排序规则MyOrdering，为降序排序
        val ts = new mutable.TreeSet[String]()(new MyOrdering())
        ts.add(nameScore)
        ts
    }

    // 合并分区中key相同的value，同时使用treeSet来进行排序
    def mergeValue(ts:mutable.TreeSet[String], nameScore:String):mutable.TreeSet[String] = {
        ts.add(nameScore)
        if(ts.size > 3) {   // 如果超过3个，删除一个再返回
            ts.dropRight(1) // scala中的集合进行操作后，本身不变，但是会返回一个新的集合
        }
        ts
    }

    // 合并不同分区中key相同的value集合，同时使用treeSet来进行排序
    def mergeCombiners(ts1:mutable.TreeSet[String], ts2:mutable.TreeSet[String]):mutable.TreeSet[String] = {
        var newTS = new mutable.TreeSet[String]()(new MyOrdering())
        // 将分区1中集合的value添加到新的treeSet中，同时进行排序和控制大小
        for(nameScore <- ts1) {
            newTS.add(nameScore)
            if(newTS.size > 3) {    // 如果数量大于3，则删除一个后再赋值给本身
                newTS = newTS.dropRight(1)
            }
        }
        // 将分区2中集合的value添加到新的treeSet中，同时进行排序和控制大小
        for(nameScore <- ts2) {
            newTS.add(nameScore)
            if(newTS.size > 3) {    // 如果数量大于3，则删除一个后再赋值给本身
                newTS = newTS.dropRight(1)
            }
        }
        newTS
    }

}

输出结果如下：

==========TopN前==========
(chinese,ls 91)
(chinese,sj 74)
(english,ww 56)
(english,ts 87)
(chinese,zs 90)
(english,ys 67)
(chinese,zl 76)
(english,mz 77)
(english,zq 88)
(chinese,yj 98)
(chinese,wb 95)
(english,gk 96)
==========TopN后==========
(english,TreeSet(gk 96, zq 88, ts 87))
(chinese,TreeSet(yj 98, wb 95, ls 91))

原文地址：http://blog.51cto.com/xpleaf/2108763

时间： 2024-11-06 20:34:03

Spark高级排序与TopN问题揭密的相关文章

spark高级排序彻底解秘

排序,真的非常重要! RDD.scala(源码) 在其,没有罗列排序,不是说它不重要! 1.基础排序算法实战 2.二次排序算法实战 3.更高级别排序算法 4.更高级别排序算法 1.基础排序算法实战启动hdfs集群 [email protected]:/usr/local/hadoop/hadoop-2.6.0$ sbin/start-dfs.sh 启动spark集群 [email protected]:/usr/local/spark/spark-1.5.2-bin-hadoop2.6$ sb

第19课：Spark高级排序彻底解密

本节课内容: 1.基础排序算法实战 2.二次排序算法实战 3.更高级别排序算法 4.排序算法内幕解密排序在Spark运用程序中使用的比较多,且维度也不一样,如二次排序,三次排序等,在机器学习算法中经常碰到,所以非常重要,必须掌握! 所谓二次排序,就是根据两列值进行排序,如下测试数据: 2 3 4 1 3 2 4 3 8 7 2 1 经过二次排序后的结果(升序): 2 1 2 3 3 2 4 1 4 3 8 7 在编写二次排序代码前,先简单的写下单个key排序的代码: val conf = ne

Spark高级排序彻底解密(DT大数据梦工厂)

内容: 1.基础排序算法实战: 2.二次排序算法实战: 3.更高局级别排序算法: 4.排序算法内幕解密: 为啥讲排序?因为在应用的时候都有排序要求. 海量数据经常排序之后要我们想要的内容. ==========基础排序算法============ scala> sc.setLogLevel("WARN") scala> val x = sc.textFile("/historyserverforSpark/README.md", 3).flatMap(_

Spark自定义排序与分区

Spark自定义排序与分区前言: 随着信息时代的不断发展,数据成了时代主题,今天的我们徜徉在数据的海洋中:由于数据的爆炸式增长,各种数据计算引擎如雨后春笋般冲击着这个时代.作为时下最主流的计算引擎之一 Spark也是从各方面向时代展示自己的强大能力.Spark无论是在数据处理还是数据分析.挖掘方面都展现出了强大的主导能力.其分布式计算能力受到越来越多的青睐.本文将介绍spark的排序以及分区. 一.Spark自定义排序在spark中定义了封装了很多高级的api,在我们的日常开发中使用这些ap

数据结构之高级排序算法

一.希尔排序希尔排序(缩小增量法) 属于插入类排序,由Shell提出,希尔排序对直接插入排序进行了简单的改进:它通过加大插入排序中元素之间的间隔,并在这些有间隔的元素中进行插入排序,从而使数据项大跨度地移动,当这些数据项排过一趟序之后,希尔排序算法减小数据项的间隔再进行排序,依次进行下去,进行这些排序时的数据项之间的间隔被称为增量,习惯上用字母h来表示这个增量. 具体代码实现: 1 package data.struct.algorithm; 2 3 //高级排序算法之希尔排序 4 class

javascript数据结构与算法--高级排序算法

高级排序算法是处理大型数据集的最高效排序算法,它是处理的数据集可以达到上百万个元素,而不仅仅是几百个或者几千个.现在我们来学习下2种高级排序算法---- 希尔排序和快速排序. 一:希尔排序: 希尔排序的核心理念是:首先比较距离较远的元素,而非相邻的元素. 基本原理:通过定义一个间隔序列来表示在排序过程中进行比较的元素之间有多远的间隔. 下面我们来看看数组[0,9,1,8,7,6,2,3,5,4] 来使用希尔排序的原理:如下图: 代码分析如下: 1. 执行 "间隔序列=3的步骤" A.

高级排序之希尔排序

希尔排序对于多达几千个数据项的,中等大小规模的数组排序表现良好,希尔排序不像快速排序和其它时间复杂度为O(n*logn)的排序算法那么快,因此,对非常大的文件排序,它不是最优选择,但是希尔排序比选择排序和插入排序这种时间复杂度为O(n²)的排序要快的多,并且它非常容易实现,代码简短希尔排序也是插入排序的一种,在插入排序中,如果最小的数在最后面,则复制的次数太多,而希尔解决了这个问题,它也是n-增量排序,它的思想是通过加大插入排序中元素的间隔,并在这些有间隔的元素中进行插入排序,当这些数据项排过

数据结构与算法之--高级排序：shell排序和快速排序【未完待续】

高级排序比简单排序要快的多,简单排序的时间复杂度是O(N^2),希尔(shell)排序的是O(N*(logN)^2),而快速排序是O(N*logN). 说明:下面以int数组的从小到大排序为例. 希尔(shell)排序希尔排序是基于插入排序的,首先回顾一下插入排序,假设插入是从左向右执行的,待插入元素的左边是有序的,且假如待插入元素比左边的都小,就需要挪动左边的所有元素,如下图所示: ==> 图1和图2:插入右边的temp柱需要outer标记位左边的五个柱子都向右挪动如图3所示,相比插入排序

处理海量数据的高级排序之——希尔排序（C++）

希尔算法简介常见排序算法一般按平均时间复杂度分为两类:O(n^2):冒泡排序.选择排序.插入排序O(nlogn):归并排序.快速排序.堆排序简单排序时间复杂度一般为O(n^2),如冒泡排序.选择排序.插入排序等高级排序时间复杂