package flow.partitioner;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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.mapred.lib.HashPartitioner;
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;

import flow.pojo.Flow;

public class FlowPartitionerMR {
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		
		conf.set("fs.defaultFS", "hdfs://hadoop01:9000");
		System.setProperty("HADOOP_USER_NAME", "hadoop");
		
		Job job = Job.getInstance(conf);
		
		job.setJarByClass(FlowPartitionerMR.class);
		
		job.setMapperClass(FlowPartitionerMRMapper.class);
		job.setMapOutputKeyClass(Flow.class);
		job.setMapOutputValueClass(NullWritable.class);
		
		job.setReducerClass(FlowPartitionerMRReducer.class);
		job.setOutputKeyClass(Flow.class);
		job.setOutputValueClass(NullWritable.class);
		
		/**
		 * 在MapReduce程式設計模型中，Partitioner的預設實現是HashPartitioner.如果
		 * 不能滿足我們的需求，按照HashPartitioner的實現方式自定義一個Partitioner元件即可
		 */
		job.setPartitionerClass(ProvincePartitioner.class);
	
//		job.setPartitionerClass(HashPartitioner.class); 
		
		
		job.setNumReduceTasks(9);
		
		Path inputPath = new Path(args[0]);
		Path outputPath = new Path(args[1]);
		FileInputFormat.setInputPaths(job, inputPath);
		FileSystem fs = FileSystem.get(conf);
		if(fs.exists(outputPath)){
			fs.delete(outputPath,true);
		}
		FileOutputFormat.setOutputPath(job,outputPath);
		
		boolean isDone = job.waitForCompletion(true);
		System.exit(isDone ? 0 : 1);
		
	}
	
	/**
	 * Mapper階段的業務邏輯
	 *
	 */
	private static class FlowPartitionerMRMapper extends Mapper<LongWritable, Text, Flow, NullWritable>{
		Flow flow = new Flow();
		/**
		 * 13480253104	2494800	2494800	4989600
		 * 
		 */

		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Flow, NullWritable>.Context context)
				throws IOException, InterruptedException {
			String[] split = value.toString().split("\t");
			String phone = split[0];
			long upFlow = Long.parseLong(split[1]);
			long downFlow = Long.parseLong(split[2]);
			long sumFlow = Long.parseLong(split[3]);
			
			/**
			 * map方法每呼叫一次，那麼就給flow物件中的對應屬性重新設定值，達到減少JVM垃圾回收的目的
			 */
			flow.setPhone(phone);
			flow.setUpFlow(upFlow);
			flow.setDownFlow(downFlow);
			flow.setSumFlow(sumFlow);
		
			/**
			 * 當前這個做法，是讓整個MapTask公用一個Flow物件。為什麼可以這麼用？
			 * 
			 * 序列化的工作機制
			 * 每次通過context.write(key,value)其實是把key和value當中的屬性值已經序列化了到其他的比如流或者記憶體或者磁碟檔案
			 * 
			 */
			context.write(flow, NullWritable.get());
			
		}
		
	}
	
	/**
	 * Reducer階段的業務邏輯
	 */
	private static class FlowPartitionerMRReducer extends Reducer<Flow, NullWritable, Flow, NullWritable>{

		@Override
		protected void reduce(Flow key, Iterable<NullWritable> values, Context context) 
				throws IOException, InterruptedException {

			/**
			 * 如果ReducerTask什麼邏輯都不用做，僅僅只是作原樣輸出：
			 * 
			 * 兩鍾實現方式:
			 * 
			 * 1、在當前的reduce方法中，直接遍歷原樣輸出
			 */
			for(NullWritable value : values){
				context.write(key, value);
			}
			
			/**
			 * 2、直接不用定義Reducer，直接使用預設實現
			 */

		}
		
		
		
	}
	
	

}
 

按照省分割槽:

package flow.partitioner;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Partitioner;

import flow.pojo.Flow;
/**
 * Partitioner的key-value的型別就是 Mapper元件的輸出的key-value的型別
 * 
 * Combiner的輸入key-value的型別也是Mapper元件的輸出key-value的型別
 * 
 * 假如Combiner的執行在Partitioner之前。
 * Partitioner的輸入的key-vlaue的型別就應該是Combiner的輸出key-value型別
 * 
 * Partitoner的執行時在Combiner之前。
 *
 */

public class ProvincePartitioner  extends Partitioner<Flow, NullWritable> {

	/**
	 * 作用：就是用來決定輸入的引數key-value到底應該進入到哪個ReducerTask
	 * 
	 * 當前自定義的Partitioner的邏輯:
	 * 	按照手機歸屬地的不同，把所有使用者的流量彙總資訊輸出到不同的結果檔案中
	 * 	如果要實現這個業務，必須依賴一張使用者的手機號和歸屬地的字典表
	 * 
	 * 	這種類似的字典表: 儲存在MySQL當中
	 * 	
	 * 	咱們做模擬實現
	 * 	假設: 134開頭的所有使用者都是北京的
	 * 		 135開頭的所有使用者都是上海的
	 * 		 ...
	 *
	 */
	@Override
	public int getPartition(Flow key, NullWritable value, int numPartitions) {
		
		return getProvinceNumber(key);
	}
	
	private static int getProvinceNumber(Flow key){
		String phone = key.getPhone();
		String phonePrefix = phone.substring(0, 3);
		
		if(phonePrefix.equals("134")){
			return 0;
		}else if (phonePrefix.equals("135")){
			return 1;
		}else if(phonePrefix.equals("136")){
			return 2;
		}else if(phonePrefix.equals("137")){
			return 3;
		}else if(phonePrefix.equals("138")){
			return 4;
		}else{
			return 5;
		}
	}
	
	

}