1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67
| public class FirstMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
private String fileName; private Text outKey = new Text(); private IntWritable outValue = new IntWritable(1);
@Override protected void setup(Context context) throws IOException, InterruptedException { FileSplit fileSplit = (FileSplit) context.getInputSplit(); fileName = fileSplit.getPath().getName(); }
@Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String[] fields = value.toString().split(" "); for (String field : fields) { String k = field + "--" + fileName; outKey.set(k); context.write(outKey, outValue); } } }
public class FirstReduce extends Reducer<Text, IntWritable, Text, IntWritable> {
private IntWritable outValue = new IntWritable();
@Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable value : values) { sum += value.get(); }
outValue.set(sum);
context.write(key, outValue); } }
public static void main(String[] args) throws InterruptedException, IOException, ClassNotFoundException { Configuration configuration = new Configuration();
Job job = Job.getInstance(configuration);
job.setJarByClass(FirstDriver.class);
job.setMapperClass(FirstMapper.class); job.setReducerClass(FirstReduce.class);
job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class);
job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class);
FileInputFormat.setInputPaths(job, new Path("D:\\dev\\hadoop\\job\\*.txt"));
FileOutputFormat.setOutputPath(job, new Path("D:\\dev\\hadoop\\job\\output"));
boolean succeed = job.waitForCompletion(true);
System.exit(succeed ? 0 : 1); }
|