很多時候需要對大檔案進行分割槽

最簡單的是ID的hash分割槽

利用MapReduce的分割槽把檔案分割成到不同的檔案中去

方便後續的計算，例如KNN可以吧預測切分成多個小片

分別讀入預測

package com.mr.partition;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
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.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 * 把預測的資料讀入記憶體然後進行迭代計算
 * 適用於預測資料很少訓練資料很多
 * 如果預測資料很多可以切分多分分別計算
 * @author lenovo
 * 1,計算歐式距離(可根據實際情況修改距離公式)
 * 2,找出最近
 *   輸出topk使用TreeSet<TopKeyWritable>自己寫TopKeyWritable排序
 */
public class IDhashMR extends Configured implements Tool {

	public static enum Counter {
		PARSER_ERR
	}

	public static class MyMap extends Mapper<LongWritable, Text, Text, Text> {
		private Text mykey = new Text();
		private Text myval = new Text();
        List  testList=new ArrayList();
		
		
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String[] array = value.toString().split(",");
				
				mykey.set(array[0]);
				myval.set(array[1]);
				context.write(mykey, myval);
		};
	}

	public static class MyReduce extends Reducer<Text, Text, Text, Text> {
		private Text val = new Text();
		Map top=new TreeMap();
		
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			// 迴圈遍歷 Interable
			int sum = 0;
			for (Text value : values) {
				// 累加
				String[] array = value.toString().split(",");
				sum += Integer.parseInt(array[0]);
			}
			val.set(sum+"");
			context.write(key, val);
		};
	}

	@Override
	public int run(String[] args) throws Exception {
		// 1 conf
		Configuration conf = new Configuration();
		conf.set("mapred.textoutputformat.separator", ",");// key value分隔符
//		DistributedCache.addCacheFile(new Path(args[2]).toUri(), conf);// 為該job新增快取檔案
		// 2 create job
		// Job job = new Job(conf, ModuleMapReduce.class.getSimpleName());
		Job job = this.parseInputAndOutput(this, conf, args);
		// 3 set job
		// 3.1 set run jar class
		// job.setJarByClass(ModuleReducer.class);
		// 3.2 set intputformat
		job.setInputFormatClass(TextInputFormat.class);
		// 3.3 set input path
		// FileInputFormat.addInputPath(job, new Path(args[0]));
		// 3.4 set mapper
		job.setMapperClass(MyMap.class);
		// 3.5 set map output key/value class
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		// 3.6 set partitioner class
		 job.setPartitionerClass(IDhashPartition.class);
		// 3.7 set reduce number
		 job.setNumReduceTasks(2);
		// 3.8 set sort comparator class
		// job.setSortComparatorClass(LongWritable.Comparator.class);
		// 3.9 set group comparator class
		// job.setGroupingComparatorClass(LongWritable.Comparator.class);
		// 3.10 set combiner class
		// job.setCombinerClass(null);
		// 3.11 set reducer class
		job.setReducerClass(MyReduce.class);
		// 3.12 set output format

		job.setOutputFormatClass(TextOutputFormat.class);
		// 3.13 job output key/value class
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		// 3.14 set job output path
		// FileOutputFormat.setOutputPath(job, new Path(args[1]));
		// 4 submit job
		boolean isSuccess = job.waitForCompletion(true);
		// 5 exit
		// System.exit(isSuccess ? 0 : 1);
		return isSuccess ? 0 : 1;
	}

	public Job parseInputAndOutput(Tool tool, Configuration conf, String[] args)
			throws Exception {
		// validate
//		if (args.length != 2) {
//			System.err.printf("Usage:%s [genneric options]<input><output>\n",
//					tool.getClass().getSimpleName());
//			ToolRunner.printGenericCommandUsage(System.err);
//			return null;
//		}
		// 2 create job
		Job job = new Job(conf, tool.getClass().getSimpleName());
		// 3.1 set run jar class
		job.setJarByClass(tool.getClass());
		// 3.3 set input path
		FileInputFormat.addInputPath(job, new Path(args[0]));
		// 3.14 set job output path
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		return job;
	}

	public static void main(String[] args) throws Exception {
		args = new String[] {
				"hdfs://192.168.192.129:9000/ml/knn/partitioner.txt",
				// "hdfs://hadoop-00:9000/home910/liyuting/output/" };
				"hdfs://192.168.192.129:9000/ml/knn/partitioner/"};
		// run mapreduce
		int status = ToolRunner.run(new IDhashMR(), args);
		// 5 exit
		System.exit(status);
	}
}
 

package com.mr.partition;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;
/**
 * 自定義分割槽
 * @author Administrator
 *
 */
public class IDhashPartition extends Partitioner<Text,Text>{
   // private static final Logger logger = LoggerFactory.getLogger(DefinedPartition.class);
    /**
    *  資料輸入來源：map輸出
    * @author zengzhaozheng
    * @param key map輸出鍵值
    * @param value map輸出value值
    * @param numPartitions 分割槽總數，即reduce task個數
    */
    public int getPartition(Text key, Text value,int numPartitions) {
        //("--------enter DefinedPartition flag--------");
        /**
        * 注意：這裡採用預設的hash分割槽實現方法
        * 根據組合鍵的第一個值作為分割槽
        * 這裡需要說明一下，如果不自定義分割槽的話，mapreduce框架會根據預設的hash分割槽方法，
        * 將整個組合將相等的分到一個分割槽中，這樣的話顯然不是我們要的效果
        */
       // ("--------out DefinedPartition flag--------");
    	// return (key.getFirstKey().hashCode()&Integer.MAX_VALUE)%numPartitions;
        return (Integer.parseInt(key.toString()))%10;
    }
}