Mapreduce setup和clearup解析_hadoop中的clearup函数

public class Matrix {
	public static class MatrixMap extends Mapper<Object,Text,Text,Text>{
		private Text map_key = new Text();
		private Text map_value = new Text();
		int columnN;
		int rowM;
		//执行map函数之前通过setup从输入文件名中取出部分信息 赋值给全局变量
		public void setup(Context context){
			Configuration conf = context.getConfiguration();
			columnN = Integer.parseInt(conf.get("columnN"));
			rowM = Integer.parseInt(conf.get("rowM"));
		}
		@Override
		protected void map(Object key, Text value,Context context)throws IOException, InterruptedException {
		FileSplit fileSplit = (FileSplit) context.getInputSplit();
		String filename = fileSplit.getPath().getName();
		if(filename.contains("M")){
			String[] tuple = value.toString().split(",");
			int i = Integer.parseInt(tuple[0]);
			String[] tuples = tuple[1].split("\t");
			int j = Integer.parseInt(tuples[0]);
			int Mij = Integer.parseInt(tuples[1]);
			for(int k =1;k<columnN+1;k++){
				map_key.set(i+","+k);
				map_value.set("M"+","+j+","+Mij);
				context.write(map_key, map_value);
			}
		}else if(filename.contains("N")){
			String[] tuple = value.toString().split(",");
			int j = Integer.parseInt(tuple[0]);
			String[] tuples = tuple[1].split("\t");
			int k = Integer.parseInt(tuples[1]);
			int Njk = Integer.parseInt(tuples[1]);
			for(int i =1;i<columnN+1;i++){
				map_key.set(i+","+k);
				map_value.set("N"+","+j+","+Njk);
				context.write(map_key, map_value);
			}
		}
		}
	}

public static void main(String[] args) throws Exception {
 
    if (args.length != 3) {
      System.err
          .println("Usage: MatrixMultiply <inputPathM> <inputPathN> <outputPath>");
      System.exit(2);
    } else {
      String[] infoTupleM = args[0].split("_");
      rowM = Integer.parseInt(infoTupleM[1]);
      columnM = Integer.parseInt(infoTupleM[2]);
      String[] infoTupleN = args[1].split("_");
      columnN = Integer.parseInt(infoTupleN[2]);
    }
 
    Configuration conf = new Configuration();
    /** 设置三个全局共享变量 **/
    conf.setInt("rowM", rowM);
    conf.setInt("columnM", columnM);
    conf.setInt("columnN", columnN);
 
    Job job = new Job(conf, "MatrixMultiply");
    job.setJarByClass(MatrixMultiply.class);
    job.setMapperClass(MatrixMapper.class);
    job.setReducerClass(MatrixReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(Text.class);
    FileInputFormat.setInputPaths(job, new Path(args[0]), new Path(args[1]));
    FileOutputFormat.setOutputPath(job, new Path(args[2]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }