二次排序原理

    二次排序就是首先按照第一字段排序，然后再对第一字段相同的行按照第二字段排序，注意不能破坏第一次排序的结果。

二次排序技术

       假设对应的Key = K有如下值：

         （K,V1)， （K,V2），…，（K,Vn）

     另外假设每个Vi是包含m个属性的一个元组，如下所示：

         （Ai1，Ai2，…，Aim）

在这里我们希望按Ai1对归约器的元组的值进行排序。我们将用R表示元组其余的属性：(Ai2，…，Aim)，因此，可以把归约器的值表示为：

        （K,(A1,R1)），（K,(A2,R2)），…，（K,(An，Rn)）

要按Ai对归约器的值进行排序，那么需要创建一个组合键：（K,Ai），新映射器将发出对应的Key=K的键值对，如下表所示：

     从上表中不难理解，定义组合键为（K,Ai），自然键为K，通过定义组合键（即为自然键增加属性Ai）。假设二次排序使用MapReduce框架对归约器的值进行排序，按照自然键（K）来完成分区。则自然键和组合键排序图如下所示：

假设有这样的一个二次排序问题的例子：考虑一个科学试验得到的温度数据，这样温度的数据如下所示（各列分别为年，月，日，温度）：

2015      1     1     10

2015      1     2     11

2015      1     3     12

2015      1     4     13

…

2015      2     1     22

2015      2     2     23

2015      2     3     24

2015      2     4     25

…

2015      3     1     20

2015      3     2     21

2015      3     3     22

2015      3     4     23

假设我们希望输出每一个[年-月]的温度，并且值按升序排序。

Spark的二次排序代码实现

    1、自定义排序分区，代码如下：

/**
  * 自定义排序分区
  **/
class SortPartitioner(partitions: Int) extends Partitioner {

    require(partitions > 0, s"分区的数量($partitions)必须大于零。")

    def numPartitions: Int = partitions

    def getPartition(key: Any): Int = key match {
        case (k: String, v: Int) => math.abs(k.hashCode % numPartitions)
        case null => 0
        case _ => math.abs(key.hashCode % numPartitions)
    }

    override def equals(other: Any): Boolean = other match {
        case o: SortPartitioner => o.numPartitions == numPartitions
        case _ => false
    }

    override def hashCode: Int = numPartitions
}

    2、二次排序代码实现

/**
  * Spark的二次排序
  **/
object SparkSecondarySort {
    def main(args: Array[String]): Unit = {
        if (args.length != 3) {
            println("输入参数<分区数> <输入路径> <输出路径>不正确")
            sys.exit(1)
        }

        //分区数量
        val partitions: Int = args(0).toInt
        //文件输入路径
        val inputPath: String = args(1)
        //文件输出路径
        val outputPath: String = args(2)
        val config: SparkConf = new SparkConf()
        config.setMaster("local[1]").setAppName("SparkSecondarySort")
        //创建Spark上下文
        val sc: SparkContext = SparkSession.builder().config(config).getOrCreate().sparkContext
        //读取文件内容
        val input: RDD[String] = sc.textFile(inputPath)
        val valueToKey: RDD[((String, Int), Int)] = input.map(x => {
            val line: Array[String] = x.split("\t")
            ((line(0) + "-" + line(1), line(3).toInt), line(3).toInt)
        })

        implicit def tupleOrderingDesc = new Ordering[Tuple2[String, Int]] {
            override def compare(x: Tuple2[String, Int], y: Tuple2[String, Int]): Int = {
                if (y._1.compare(x._1) == 0) -y._2.compare(x._2)
                else -y._1.compare(x._1)
            }
        }

        val sorted: RDD[((String, Int), Int)] = valueToKey.repartitionAndSortWithinPartitions(new SortPartitioner(partitions))
        val result = sorted.map {
            case (k, v) => (k._1, v.toString())
        }.reduceByKey(_ + "," + _)
        result.saveAsTextFile(outputPath)
        // done
        sc.stop()
    }
}

运行结果：

(2015-1,5,6,7,8,9,10,10,11,11,12,12,13,13,14,14,15,15,16,16,17,17,18,18,19,19,20,20,21,21,22,22)
(2015-3,18,19,20,20,20,21,21,21,22,22,22,23,23,23,24,24,24,25,25,25,26,26,26,27,27,27,28,28,28,29,30)
(2015-2,12,13,14,15,16,17,18,19,20,21,22,22,23,23,24,24,25,25,26,26,27,28,29,30,30,30,31,32)

想弄懂hadoop的二级排序，需要对hadoop的shufle过程非常清楚。

（注：在reduce接受数据阶段，setSortComparatorClass是对key进行的排序处理，setGroupingComparatorClass是对相同值的value进行排序处理）

下面是一个hadoop的二级排序案例：

编写mapReduce程序，将下面的左边内容进行排序，排序后变成右边的样子。

mapReduce的程序如下：

第一步：定义一个组合键的bean

package com.bigdata.demo15_two_class_paixu;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.WritableComparable;
/**
 * 自定义组合键，用于map阶段的sort小阶段
 * @author Administrator
 * 2018年5月31日上午8:16:38
 */
public class CombinationKey implements WritableComparable<CombinationKey>{

	private String firstKey;
	private Integer secondKey;
	public String getFirstKey() {
		return firstKey;
	}

	public void setFirstKey(String firstKey) {
		this.firstKey = firstKey;
	}

	public Integer getSecondKey() {
		return secondKey;
	}

	public void setSecondKey(Integer secondKey) {
		this.secondKey = secondKey;
	}

	public void write(DataOutput out) throws IOException {
		out.writeUTF(this.firstKey);
		out.writeInt(this.secondKey);
	}

	public void readFields(DataInput in) throws IOException {
		this.firstKey=in.readUTF();
		this.secondKey=in.readInt();
	}

	public int compareTo(CombinationKey o) {
		return this.firstKey.compareTo(o.getFirstKey());
	}

	
}

package com.bigdata.demo15_two_class_paixu;

import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
/**
 * 自定义比较器
 * @author Administrator
 * 2018年5月31日上午8:40:58
 */
public class DefineCompparator extends WritableComparator{

	protected DefineCompparator() {
		super(CombinationKey.class,true);
	}
	@Override
	public int compare(WritableComparable a, WritableComparable b) {
		
		CombinationKey ck1=(CombinationKey) a;
		CombinationKey ck2=(CombinationKey) b;
		int cp1 = ck1.getFirstKey().compareTo(ck2.getFirstKey());
		if(cp1!=0) {
			//结束排序
			return cp1;
		}else {
			return  ck1.getSecondKey()-ck2.getSecondKey();
		}
	}
}

第三步：自定义一个分区器，在shuffle阶段使用

package com.bigdata.demo15_two_class_paixu;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Partitioner;
/**
 * 自定义分区
 * @author Administrator
 * 2018年5月31日上午8:20:58
 */
public class DefinedPartition extends Partitioner<CombinationKey, IntWritable>{
	/**
	 * @param key map输出，这里根据组合键的第一个值进行分区
	 * @param value map输出的key
	 * @param numPartitions 分区总数，即reduce的个数
	 */
	@Override
	public int getPartition(CombinationKey key, IntWritable value, int numPartitions) {
		return (key.getFirstKey().hashCode()&Integer.MAX_VALUE)%numPartitions;
	}

}

第四步：自定义一个分区器，在shuffle阶段使用

package com.bigdata.demo15_two_class_paixu;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Partitioner;
/**
 * 自定义分区
 * @author Administrator
 * 2018年5月31日上午8:20:58
 */
public class DefinedPartition extends Partitioner<CombinationKey, IntWritable>{
	/**
	 * @param key map输出，这里根据组合键的第一个值进行分区
	 * @param value map输出的key
	 * @param numPartitions 分区总数，即reduce的个数
	 */
	@Override
	public int getPartition(CombinationKey key, IntWritable value, int numPartitions) {
		return (key.getFirstKey().hashCode()&Integer.MAX_VALUE)%numPartitions;
	}

}

第五步：编写mapReduce程序

package com.bigdata.demo15_two_class_paixu;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class SecondSortMapReduce {
	/**
	 * 使用内部类的形式，定义mapper程序
	 * @author Administrator
	 * 2018年5月31日上午11:06:30
	 */
	static class SecondSortMapper extends Mapper<LongWritable, Text, CombinationKey, IntWritable>{
		String[] split=null;
		CombinationKey kv=new CombinationKey();
		IntWritable v=new IntWritable();
		@Override
		protected void map(LongWritable key, Text value,
				Context context)
				throws IOException, InterruptedException {
			split = value.toString().split(" ");
			kv.setFirstKey(split[0]);
			int vv = Integer.parseInt(split[1]);
			v.set(vv);
			kv.setSecondKey(vv);
			context.write(kv, v);
		}
	}
	/**
	 * 使用内部类的形式，定义reduce程序
	 * @author Administrator
	 * 2018年5月31日上午11:06:51
	 */
	static class SecondSortReducer extends Reducer<CombinationKey, IntWritable, Text, Text>{
		Text k=new Text();
		Text v=new Text();
		@Override
		protected void reduce(CombinationKey first_second, Iterable<IntWritable> seconds,
				Context context)
				throws IOException, InterruptedException {
			StringBuilder sb=new StringBuilder();
			for(IntWritable second:seconds) {
				sb.append(second.get()+",");
			}
			k.set(first_second.getFirstKey());
			v.set(sb.toString().substring(0, sb.toString().length()-1));
			context.write(k, v);
		}
	}
	/**
	 * 主函数
	 * @param args
	 * @throws IOException
	 * @throws ClassNotFoundException
	 * @throws InterruptedException
	 */
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		Configuration conf=new Configuration();
		Job job = Job.getInstance(conf);
		
		job.setJarByClass(SecondSortMapReduce.class);
		job.setMapperClass(SecondSortMapper.class);
		job.setReducerClass(SecondSortReducer.class);
		
		//设置分区和reduce数目
		job.setPartitionerClass(DefinedPartition.class);
		job.setNumReduceTasks(1);
		//设置自定义的分组策略
		job.setGroupingComparatorClass(DefinedGroupSort.class);
		//设置自定义的比较策略
		job.setSortComparatorClass(DefineCompparator.class);
		
		job.setMapOutputKeyClass(CombinationKey.class);
		job.setMapOutputValueClass(IntWritable.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		//设置输入数据
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
		boolean res = job.waitForCompletion(true);
		
		System.exit(res?0:1);
		
	}
}

第六步：在hadoop集群上运行

将程序打包，放到hadoop集群，运行

[hadoop@mini02 ~]$ hadoop jar \
> ./jars/Review06_hdfs-0.0.1-SNAPSHOT.jar \
> com.bigdata.demo15_two_class_paixu.SecondSortMapReduce \
> /paixu/input01 \
> /paixu/output03

[hadoop@mini03 ~]$ hdfs dfs -cat /paixu/output02/part-r-00000
hadoop  23,32,342
hive    12,42,204,2345
spark   3,16,349

　　有时我们因工作原因需要进行excel时间排序。

　　第一，excel时间排序：按年排序

　　如下面的截图，需要对B列的时间按年排序。

　　方法很简单，单击数据——排序，弹出“排序”对话框，按“出生日期”升序排序就完成了excel时间排序：按年排序。

　　第二，excel时间排序：按月排序

　　还是以上面的截图为例，如果需要完成excel时间排序：按月排序，我们在C列添加一个辅助列，C2输入公式：=TEXT(B2,"mmdd")，下拉提取B列日期的月和日，然后再以“辅助列”为主要关键字排序即可完成excel时间排序：按月排序。