MapReduce-DataClean/DataClean.java at main · alvintoto/MapReduce-DataClean · 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
import java.io.IOException;
import java.util.StringTokenizer;

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

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

        private int counter = 1;
        private String prev_value = "";
        private IntWritable output_key = new IntWritable();
        private Text output_val = new Text();

        public void map(Object key, Text value, Context context
        ) throws IOException, InterruptedException {
            StringTokenizer itr = new StringTokenizer(value.toString());
            while (itr.hasMoreTokens()) {
                String[] key_val_pair = itr.nextToken().split(",");

                int temp_key = Integer.parseInt(key_val_pair[0]);
                String temp_val = key_val_pair[1];

                if (counter == temp_key) {
                    output_key.set(temp_key);
                    output_val.set(temp_val);
                    context.write(output_key, output_val);

                }
                else {
                    output_key.set(counter - 1);
                    output_val.set(prev_value);
                    context.write(output_key, output_val);

                    output_key.set(temp_key);
                    output_val.set(temp_val);
                    context.write(output_key, output_val);

                    counter += 1;
                }

                counter += 1;
                prev_value = temp_val;

            }
        }
    }

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

        private int counter;
        private IntWritable output_key = new IntWritable();
        private Text output_val = new Text();

        public void reduce(IntWritable key, Iterable<Text> values,
                           Context context
        ) throws IOException, InterruptedException {
            counter = key.get();
            for (Text val : values) {
                output_key.set(counter);
                output_val.set(val);
                context.write(output_key, output_val);
                counter += 1;
            }
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "data clean");
        job.setJarByClass(DataClean.class);
        job.setMapperClass(DataCleanMapper.class);
        job.setCombinerClass(DataCleanReducer.class);
        job.setReducerClass(DataCleanReducer.class);
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(Text.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}