Basic-MapReduce/Filter.java at master · novamind/Basic-MapReduce · GitHub

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
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.IntWritable;
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.output.FileOutputFormat;

public class Filter {


//	public static class FilterMapper extends Mapper<Object, Text, Object, Text> {
//
//		@Override
//		protected void map(Object key, Text value, Context context) throws IOException, InterruptedException{
//
//			String filterValue = context.getConfiguration().get("filtervalue");
//
//			if(value.toString().contains(filterValue))
//				context.write(key,value);
//
//		}
//
//
//	}

	public static class FilterMapper extends Mapper<Object, Text, Text, IntWritable> {

		private final static IntWritable ONE = new IntWritable(1);

		@Override
		protected void map(Object key, Text value, Mapper<Object,Text,Text,IntWritable>.Context context) throws IOException, InterruptedException{

			String keyword = context.getConfiguration().get("filterkeyword");
			System.out.println(keyword);

			if(value.toString().contains(keyword)){
				Text newKey = new Text(keyword);
				context.write(newKey,ONE);
			}

		}


	}

	public static class FilterReducer extends Reducer<Text, IntWritable, Text, IntWritable>{

		private IntWritable result = new IntWritable();

		@Override
		protected void reduce(Text key, Iterable<IntWritable> values,
				Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {


			int sum = 0;

			for(IntWritable val : values){
				sum  += val.get();
			}


			result.set(sum);

			context.write(key,result);

		}
	}


	public static void main(String[] args) throws Exception {

		Configuration conf = new Configuration();


		conf.set("filterkeyword",args[2]);

		Job job = Job.getInstance(conf, "filter");

		job.setJarByClass(Filter.class);
		job.setMapperClass(FilterMapper.class);
		job.setReducerClass(FilterReducer.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);

		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		System.exit(job.waitForCompletion(true) ? 0 : 1);

	}

}