1. 分析需求(首先对第一个字段排序,然后在对第二个字段排序)

    杂乱的原始数据           排序完成的数据

        a,1                     a,1

        b,1                     a,2

        a,2       [排序]        a,100

        b,6        ===>         b,-3

        c,2                     b,-2

        b,-2                    b,1

        a,100                   b,6

        b,-3                    c,-7

        c,-7                    c,2

2. 分析[MapRedice过程]

     1> 分析数据传入通过input()传入map()

     2> map()对数据进行层层过滤,以达到我们想要的数据源,

     3> 过滤方法中可添加自定义计数器

     4> 过滤后写入context,转入shuffle阶段

     5> 可以说大部分shuffle阶段是map()端的shuffle

     6> 具体shullfe中,数据经过默认分区(hashPartitioner),而默认分区规则是获取

        (key.getHashCode() & Integer.MAX_VALUE)%numReudeceTasks;当然默认reduce数目就一个,

        reduce输出的文件也就一个,我是这样认为的,经过输出测试,就算你设置了自定义的分区,但你的partition数目

        并没设置,仍然走默认分区

     7> 分区之后对是分区的一个排序,再对分区中的数据进行排序,排序规则按照key排序,我们可以自定义数据类型对其

        设置排序规则,比如二次排序,可以自定义一个组合的key,在组合key中定义根据第一个字段排序,如果第一个字段

        相同,那么再进行对第二个字段排序,以达到二次排序的目的,在分区排序后进入分组阶段也是默认按照key分的,

        分组需要实现RawComparator

     8> 分组之后是merge个并归排序然后进入reduce,其中分组决定数据进入某个reduce,而分区决定了reduce阶段生成

        文件的数目,分组算是shuffle阶段对程序运行的一个优化吧我是这么理解的

 3. 分析[二次排序]

     1> 从上面的数据可以看出,我们可以自定义一个数据类型,来存放第一个和第二个字段,然后自定义一个比较器来

        说明排序规则按照key中的第一个字段进行排序,这里涉及到自定义数据需要实现WritableComparable也可以

        分别继承Writable和Comparable,反正越方便越好

     2> 接下来看看分区操作,该例只生成一个排好序的文件,不用自定义分区,自定义分区后也不会走该类,自定义分区需要

        继承Partitioner,注意是继承,我们自己要重写分区规则

     3> 然后是分组操作,分组为优化考虑还是有必要的,我们设计分组规则为按照自定义数据类型的第一个字段进行分组,

        分组需要实现RawComparator

     4> 考虑哪里还需要优化,根据数据源的数据量,字段是否必在,长度情况,

        类型情况,是否使用combine与自定义压缩类,数值为负数等,在比较器中既然定义了根据第二个字段比较,我想也

        没必要加个大数减个大数

　　效果展示:

    　　数据源      map()后         shuffle阶段后         reduce()后

     　　a,1        a#1,1          a#1 [1,2,100]          a   1

     　　b,1        b#1,1          b#-3 [-3,-2,1,6]       a   2

     　　a,2        a#2,2          c#-7 [-7,2]            a   100

     　　b,6        b#6,6                                 b   -3

     　　c,2        c#2,2                                 b   -2

     　　b,-2       b#-2,-2                               b   1

     　　a,100      a#100,100                             b   6

     　　b,-3       b#-3,-3                               c   -7

     　　c,-7       c#-7,-7                               c   2

SSortMr.java ## 主类

============

package com.bigdata_senior.SSortMr;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

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 SSortMr {

	//Mapper Class

	private static class SSortMapper

	    extends Mapper<LongWritable, Text, SecondaryWritable, LongWritable>{

		private SecondaryWritable mapOutKey = new SecondaryWritable();

		private LongWritable mapOutValue = new LongWritable();

		@Override

		public void map(LongWritable key, Text value, Context context)

				throws IOException, InterruptedException {

			String lineValue = value.toString();

			String [] strValue = lineValue.split(",");

			mapOutKey.set(strValue[0],Integer.valueOf(strValue[1]));

			mapOutValue.set(Integer.valueOf(strValue[1]));

			context.write(mapOutKey, mapOutValue);

			System.out.println("key-->"+mapOutKey+" value-->"+mapOutValue);

		}

	}

	//Reduce Class

	private static class SSortReduce

	    extends Reducer<SecondaryWritable, LongWritable, Text, LongWritable>{

		private Text reduceOutKey = new Text();

		@Override

		public void reduce(SecondaryWritable key, Iterable<LongWritable> values,Context context)

				throws IOException, InterruptedException {

			for(LongWritable value : values){

				reduceOutKey.set(key.getFirst()+"#"+key.getSecond());

				context.write(reduceOutKey, value);

			}

		}

	}

	//Driver

	public int run(String[] args) throws Exception {

		Configuration configuration = new Configuration();

		Job job = Job.getInstance(configuration, this.getClass().getSimpleName());

		job.setJarByClass(this.getClass());

		//job.setNumReduceTasks(3);

		//input

		Path inPath = new Path(args[0]);

		FileInputFormat.addInputPath(job,inPath);

		//output

		Path outPath = new Path(args[1]);

		FileOutputFormat.setOutputPath(job, outPath);

		//mapper

		job.setMapperClass(SSortMapper.class);

		job.setMapOutputKeyClass(SecondaryWritable.class);

		job.setMapOutputValueClass(LongWritable.class);

		//partitioner

		//job.setPartitionerClass(SecondaryPartionerCLass.class);

		//group

		job.setGroupingComparatorClass(SecondaryGroupClass.class);

		//Reduce

		job.setReducerClass(SSortReduce.class);

		job.setOutputKeyClass(Text.class);

		job.setOutputValueClass(LongWritable.class);

		//submit job

		boolean isSuccess = job.waitForCompletion(true);

		return isSuccess ? 0 : 1;

	}

	public static void main(String[] args) throws Exception {

		args = new String[]{

			"hdfs://hadoop09-linux-01.ibeifeng.com:8020/user/liuwl/tmp/sortmr/input",

			"hdfs://hadoop09-linux-01.ibeifeng.com:8020/user/liuwl/tmp/sortmr/output13"

		};

		//run job

		int status = new SSortMr().run(args);

		System.exit(status);

	}

}

SecondaryWritable.java ## 自定义数据类型

======================

package com.bigdata_senior.SSortMr;

import java.io.DataInput;

import java.io.DataOutput;

import java.io.IOException;

import org.apache.hadoop.io.WritableComparable;

public class SecondaryWritable implements WritableComparable<SecondaryWritable> {

	private String first;

	private int second;

	public SecondaryWritable() {}

	public SecondaryWritable(String first,int second){

		this.set(first, second);

	}

	public void set(String fist,int second){

		this.first = fist;

		this.second = second;

	}

	public String getFirst() {

		return first;

	}

	public void setFirst(String first) {

		this.first = first;

	}

	public int getSecond() {

		return second ;

	}

	public void setSecond(int second) {

		this.second = second ;

	}

	@Override

	public void write(DataOutput out) throws IOException {

		out.writeUTF(this.first);

		out.writeInt(this.second);

	}

	@Override

	public void readFields(DataInput in) throws IOException {

		this.first = in.readUTF();

		this.second = in.readInt();

	}

	@Override

	public int compareTo(SecondaryWritable o) {

		int comp = this.first.compareTo(o.first);

		if(0 != comp){

			return comp;

		}

		return Integer.valueOf(this.second).compareTo(Integer.valueOf(o.second));

	}

	@Override

	public String toString() {

		return first + "#" + second;

	}

	@Override

	public int hashCode() {

		final int prime = 31;

		int result = 1;

		result = prime * result + ((first == null) ? 0 : first.hashCode());

		result = prime * result + second;

		return result;

	}

	@Override

	public boolean equals(Object obj) {

		if (this == obj)

			return true;

		if (obj == null)

			return false;

		if (getClass() != obj.getClass())

			return false;

		SecondaryWritable other = (SecondaryWritable) obj;

		if (first == null) {

			if (other.first != null)

				return false;

		} else if (!first.equals(other.first))

			return false;

		if (second != other.second)

			return false;

		return true;

	}

}

SecondaryPartionerCLass.java ## 自定义分区规则(已注释不用)

============================

package com.bigdata_senior.SSortMr;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.mapreduce.Partitioner;

public class SecondaryPartionerCLass extends Partitioner<SecondaryWritable, LongWritable> {

	@Override

	public int getPartition(SecondaryWritable key, LongWritable value,

			int numPartitions) {

		return (key.getFirst().hashCode() & Integer.MAX_VALUE) % numPartitions;

	}

}

SecondaryGroupClass.java ## 自定义分组规则

========================

package com.bigdata_senior.SSortMr;

import java.util.Arrays;

import org.apache.hadoop.io.RawComparator;

import org.apache.hadoop.io.WritableComparator;

public class SecondaryGroupClass implements RawComparator<SecondaryWritable> {

	@Override

	public int compare(SecondaryWritable o1, SecondaryWritable o2) {

		System.out.println("o1: "+o1.toString()+" o2: "+o2.toString());

		return o1.getFirst().compareTo(o2.getFirst());

	}

	@Override

	public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {

		System.out.println("b1: "+Arrays.toString(b1)+" b2: "+Arrays.toString(b2));

		return WritableComparator.compareBytes(b1, 0, l1-4, b2, 0, l2-4);

	}

}

另外还可以: ## 但这个对于小数据可用，大数据将非常消耗资源

SSortMr2.java

=============

package com.bigdata_senior.SSortMr2;

import java.io.IOException;

import java.util.ArrayList;

import java.util.Arrays;

import java.util.Collection;

import java.util.Collections;

import java.util.Iterator;

import java.util.List;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

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 SSortMr2 {

	//Mapper Class

	private static class SSortMapper extends Mapper<LongWritable, Text, Text, LongWritable>{

		private Text mapOutKey = new Text();

		private LongWritable mapOutValue = new LongWritable();

		@Override

		public void map(LongWritable key, Text value, Context context)

				throws IOException, InterruptedException {

			String lineValue = value.toString();

			String [] strValue = lineValue.split(",");

			mapOutKey.set(strValue[0]);

			mapOutValue.set(Integer.valueOf(strValue[1]));

			context.write(mapOutKey, mapOutValue);

			System.out.println("key-->"+mapOutKey+" value-->"+mapOutValue);

		}

	}

	//Reduce Class

	private static class SSortReduce extends Reducer<Text, LongWritable, Text, Long>{

		@Override

		public void reduce(Text key, Iterable<LongWritable> values,Context context)

				throws IOException, InterruptedException {

			List<Long> longList = new ArrayList<Long>();

			for(LongWritable value: values){

				longList.add(value.get());

			}

			Collections.sort(longList);

			for(Long value : longList){

				System.out.println("key--> "+key+" value--> "+value);

				context.write(key, value);

			}

		}

	}

	//Driver

	public int run(String[] args) throws Exception {

		Configuration configuration = new Configuration();

		Job job = Job.getInstance(configuration, this.getClass().getSimpleName());

		job.setJarByClass(this.getClass());

		//input

		Path inPath = new Path(args[0]);

		FileInputFormat.addInputPath(job,inPath);

		//output

		Path outPath = new Path(args[1]);

		FileOutputFormat.setOutputPath(job, outPath);

		//mapper

		job.setMapperClass(SSortMapper.class);

		job.setMapOutputKeyClass(Text.class);

		job.setMapOutputValueClass(LongWritable.class);

		//Reduce

		job.setReducerClass(SSortReduce.class);

		job.setOutputKeyClass(Text.class);

		job.setOutputValueClass(Long.class);

		//submit job

		boolean isSuccess = job.waitForCompletion(true);

		return isSuccess ? 0 : 1;

	}

	public static void main(String[] args) throws Exception {

		args = new String[]{

			"hdfs://hadoop09-linux-01.ibeifeng.com:8020/user/liuwl/tmp/sortmr/input",

			"hdfs://hadoop09-linux-01.ibeifeng.com:8020/user/liuwl/tmp/sortmr/output22"

		};

		//run job

		int status = new SSortMr2().run(args);

		System.exit(status);

	}

}

1. join(组合)

2. 即两张或两张以上的数据源数据组合输出

3. 由于学了hive,感觉MapReduce的join不再是重点,因为在MapReduce处理

    1> 为止join表数目

    2> 操作繁琐,过滤多样,可能会考虑不全

    3> 资源消耗较重

4. MapReduce的join大致就是将两张表加载进内存,在数据混淆情况下,为其设置自定义数据类型以区分两张表,

   然后在reudece()中分别获取表并指定输出结果,当然处理join的方式还有很多,比如setup()加载一张表存进集合处理

JoinMr.java ## 主类

===========

package com.bigdata_senior.joinMr;

import java.io.IOException;

import java.util.ArrayList;

import java.util.List;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.NullWritable;

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 JoinMr {

	//Mapper Class

	private static class WordCountMapper extends

		Mapper<LongWritable, Text, LongWritable, JoinWritable>{

		private LongWritable mapoutputkey = new LongWritable();

		private JoinWritable mapoutputvalue = new JoinWritable();

		@Override

		protected void setup(Context context) throws IOException,

				InterruptedException {

		}

		@Override

		public void map(LongWritable key, Text value, Context context)

				throws IOException, InterruptedException {

			String lineValue = value.toString();

			String [] strValue = lineValue.split(",");

			int length = strValue.length;

			if(3 != length && 4 != length){

				return;

			}

			//get cid

			Long cid = Long.valueOf(strValue[0]);

			//get cname

			String cname = strValue[1];

			//set customer

			if(3 == length){

				String phone = strValue[2];

				mapoutputkey.set(cid);

				mapoutputvalue.set("customer", cname + "," + phone);

			}

			//set order

			if(4 == length){

				String price = strValue[2];

				String date = strValue[3];

				mapoutputkey.set(cid);

				mapoutputvalue.set("order", cname +","+price +","+ date);

			}

			context.write(mapoutputkey, mapoutputvalue);

		}

	}

	//Reduce Class

	private static class WordCountReduce extends

		Reducer<LongWritable, JoinWritable, NullWritable, Text>{

		private Text outputValue = new Text();

		@Override

		public void reduce(LongWritable key, Iterable<JoinWritable> values,Context context)

				throws IOException, InterruptedException {

			String customerInfo = null;

			List<String> orderList = new ArrayList<String>();

			for(JoinWritable value : values){

				if("customer".equals(value.getTag())){

					customerInfo = value.getData();

					System.out.println(customerInfo);

				}else if("order".equals(value.getTag())){

					orderList.add(value.getData());

				}

			}

			for(String order: orderList){

				outputValue.set(key.get()+","+customerInfo+","+order);

				context.write(NullWritable.get(), outputValue);

			}

		}

	}

	//Driver

	public int run(String[] args) throws Exception {

		Configuration configuration = new Configuration();

		Job job = Job.getInstance(configuration, this.getClass().getSimpleName());

		job.setJarByClass(this.getClass());

		//input

		Path inPath = new Path(args[0]);

		FileInputFormat.addInputPath(job,inPath);

		//output

		Path outPath = new Path(args[1]);

		FileOutputFormat.setOutputPath(job, outPath);

		//mapper

		job.setMapperClass(WordCountMapper.class);

		job.setMapOutputKeyClass(LongWritable.class);

		job.setMapOutputValueClass(JoinWritable.class);

		//Reduce

		job.setReducerClass(WordCountReduce.class);

		job.setOutputKeyClass(NullWritable.class);

		job.setOutputValueClass(Text.class);

		//submit job

		boolean isSuccess = job.waitForCompletion(true);

		return isSuccess ? 0 : 1;

	}

	public static void main(String[] args) throws Exception {

		args = new String[]{

			"hdfs://hadoop09-linux-01.ibeifeng.com:8020/user/liuwl/tmp/join/input",

			"hdfs://hadoop09-linux-01.ibeifeng.com:8020/user/liuwl/tmp/join/output2"

		};

		//run job

		int status = new JoinMr().run(args);

		System.exit(status);

	}

}

JoinWritable.java ## 自定义数据类型

package com.bigdata_senior.joinMr;

import java.io.DataInput;

import java.io.DataOutput;

import java.io.IOException;

import org.apache.hadoop.io.Writable;

public class JoinWritable implements Writable {

	private String tag;

	private String data;

	public JoinWritable(){}

	public JoinWritable(String tag,String data){

		this.set(tag, data);

	}

	public void set(String tag,String data){

		this.setTag(tag);

		this.setData(data);

	}

	public String getTag() {

		return tag;

	}

	public void setTag(String tag) {

		this.tag = tag;

	}

	public String getData() {

		return data;

	}

	public void setData(String data) {

		this.data = data;

	}

	@Override

	public void write(DataOutput out) throws IOException {

		out.writeUTF(this.getTag());

		out.writeUTF(this.getData());

	}

	@Override

	public void readFields(DataInput in) throws IOException {

		this.setTag(in.readUTF());

		this.setData(in.readUTF());

	}

	@Override

	public int hashCode() {

		final int prime = 31;

		int result = 1;

		result = prime * result + ((data == null) ? 0 : data.hashCode());

		result = prime * result + ((tag == null) ? 0 : tag.hashCode());

		return result;

	}

	@Override

	public boolean equals(Object obj) {

		if (this == obj)

			return true;

		if (obj == null)

			return false;

		if (getClass() != obj.getClass())

			return false;

		JoinWritable other = (JoinWritable) obj;

		if (data == null) {

			if (other.data != null)

				return false;

		} else if (!data.equals(other.data))

			return false;

		if (tag == null) {

			if (other.tag != null)

				return false;

		} else if (!tag.equals(other.tag))

			return false;

		return true;

	}

	@Override

	public String toString() {

		return tag + "," +data;

	}

}