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

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);
}
}