HDPCD-Java-复习笔记（17） - lab

迈不过友情╰ 2022-06-07 09:43 119阅读 0赞

**Java lab booklet**

**MRUnit Test**

package average;
    
    import java.io.IOException;
    import java.util.ArrayList;
    import java.util.List;
    
    import org.apache.hadoop.io.DoubleWritable;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mrunit.mapreduce.MapDriver;
    import org.apache.hadoop.mrunit.mapreduce.MapReduceDriver;
    import org.apache.hadoop.mrunit.mapreduce.ReduceDriver;
    import org.junit.Before;
    import org.junit.Test;
    
    import average.AverageJob.AverageMapper;
    import average.AverageJob.AverageReducer;
    
    public class AverageJobTest {
    	private MapDriver<LongWritable, Text, Text, Text> mapDriver;
    	private ReduceDriver<Text, Text, Text, DoubleWritable> reduceDriver;
    	private MapReduceDriver<LongWritable, Text, Text, Text, Text, DoubleWritable> mapReduceDriver;
    	
    	@Before
    	public void setup(){
    		AverageMapper mapper = new AverageMapper();
    		AverageReducer reducer = new AverageReducer();
    		mapDriver = MapDriver.newMapDriver(mapper);
    	    reduceDriver = ReduceDriver.newReduceDriver(reducer);
    	    mapReduceDriver = MapReduceDriver.newMapReduceDriver(mapper, reducer);
    	}
    	
    	@Test
    	public void testMapper(){
    		LongWritable inputKey = new LongWritable(0);
    		Text inputValue = new Text("Abbeville, SC,45001,6581,7471, 6787,195278,302280,29673,40460,3042,3294");
    		mapDriver.withInput(inputKey, inputValue);
    		Text outputKey = new Text("SC");
    		Text outputValue = new Text("40460,1");
    		mapDriver.withOutput(outputKey, outputValue);
    		try {
    			mapDriver.runTest();
    		} catch (IOException e) {
    			e.printStackTrace();
    		}
    	}
    	
    	@Test
    	public void testReducer() throws IOException{
    		Text inputKey = new Text("SC");
    		Text inputValue1 = new Text("40468,4");
    		Text inputValue2 = new Text("40462,8");
    		List<Text> values = new ArrayList<Text>();
    		values.add(inputValue1);
    		values.add(inputValue2);
    		reduceDriver.withInput(inputKey, values);
    		
    		Text outputKey = new Text("SC");
    		DoubleWritable outputValue = new DoubleWritable(6744.166666666667);
    		reduceDriver.withOutput(outputKey, outputValue);
    		reduceDriver.runTest();
    	}
    	
    	@Test
    	public void testMapperReducer() throws IOException{
    		LongWritable inputKey = new LongWritable(0);
    		Text inputValue = new Text("Abbeville, SC,45001,6581,7471, 6787,195278,302280,29673,40460,3042,3294");
    		mapReduceDriver.withInput(inputKey, inputValue);
    		Text outputKey = new Text("SC");
    		DoubleWritable outputValue = new DoubleWritable(40460.0);
    		mapReduceDriver.withOutput(outputKey, outputValue);
    		mapReduceDriver.runTest();
    	}
    	
    }

package average;
    
    import java.io.IOException;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.conf.Configured;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.DoubleWritable;
    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.Partitioner;
    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;
    
    public class AverageJob extends Configured implements Tool {
    	
    	public static class AveragePartitioner extends Partitioner<Text, Text> {
    
    		@Override
    		public int getPartition(Text key, Text value, int numPartitions) {
    			char firstLetter = key.toString().trim().charAt(0);
    			return firstLetter % numPartitions;
    		}
    	}
    
    	public enum Counters {MAP, COMBINE, REDUCE} 
    
    	public static class AverageMapper extends Mapper<LongWritable, Text, Text, Text> {
    		private Text outputKey = new Text();
    		private Text outputValue = new Text();
    		
    		@Override
    		protected void map(LongWritable key, Text value, Context context)
    				throws IOException, InterruptedException {
    			String [] values = value.toString().split(",");
    			outputKey.set(values[1].trim());
    			outputValue.set(values[9].trim() + ",1");
    			context.write(outputKey, outputValue);
    			context.getCounter(Counters.MAP).increment(1);
    		}
    
    		@Override
    		protected void cleanup(Context context)
    				throws IOException, InterruptedException {
    			System.out.println("MAP counter = " + context.getCounter(Counters.MAP).getValue());
    		}
    	}
    
    	public static class AverageCombiner extends Reducer<Text, Text, Text, Text> {
    		private Text outputValue = new Text();
    		
    		@Override
    		protected void reduce(Text key, Iterable<Text> values, Context context)
    				throws IOException, InterruptedException {
    			long sum = 0;
    			int count = 0;
    			while(values.iterator().hasNext()) {
    				String [] current = values.iterator().next().toString().split(",");
    				sum += Integer.parseInt(current[0]);
    				count += Integer.parseInt(current[1]);
    			}
    			outputValue.set(sum + "," + count);
    			context.write(key, outputValue);
    			context.getCounter(Counters.COMBINE).increment(1);
    		}		
    
    		@Override
    		protected void cleanup(Context context)
    				throws IOException, InterruptedException {
    			System.out.println("COMBINE counter = " + context.getCounter(Counters.COMBINE).getValue());
    		}
    	}
    
    	public static class AverageReducer extends Reducer<Text, Text, Text, DoubleWritable> {
    		DoubleWritable outputValue = new DoubleWritable();
    		
    		@Override
    		protected void reduce(Text key, Iterable<Text> values, Context context)
    				throws IOException, InterruptedException {
    			double sum = 0.0;
    			int count = 0;
    			while(values.iterator().hasNext()) {
    				String [] current = values.iterator().next().toString().split(",");
    				sum += Long.parseLong(current[0]);
    				count += Integer.parseInt(current[1]);
    			}
    			outputValue.set(sum/count);
    			context.write(key, outputValue);
    			context.getCounter(Counters.REDUCE).increment(1);
    		}
    
    		@Override
    		protected void cleanup(Context context)
    				throws IOException, InterruptedException {
    			System.out.println(context.getCounter(Counters.REDUCE).getValue());
    		}
    	}	
    
    	@Override
    	public int run(String[] arg0) throws Exception {
    		Configuration conf = super.getConf();
    		Job job = Job.getInstance(conf, "AverageJob");
    		job.setJarByClass(AverageJob.class);
    
    		Path out = new Path("counties/output");
    		out.getFileSystem(conf).delete(out, true);
    		FileInputFormat.setInputPaths(job, "counties");
    		FileOutputFormat.setOutputPath(job, out);
    		
    
    		job.setMapperClass(AverageMapper.class);
    		job.setReducerClass(AverageReducer.class);
    		job.setCombinerClass(AverageCombiner.class);
    		job.setPartitionerClass(AveragePartitioner.class);
    		job.setInputFormatClass(TextInputFormat.class);
    		job.setOutputFormatClass(TextOutputFormat.class);
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(Text.class);
    		job.setNumReduceTasks(5);
    
    		return job.waitForCompletion(true)?0:1;
    	}
    
    
    	public static void main(String[] args) {
    		int result = 0;
    		try {
    			result = ToolRunner.run(new Configuration(),  new AverageJob(), args);
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    		System.exit(result);
    	}
    }