Custom Partioner- Store the Special Character key in one reducer and normal key in another reducer

FileLineWritable.java
package com.srini.inputformat;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;

public class FileLineWritable implements WritableComparable<FileLineWritable> {
    public long offset;
    public String fileName;

    public void readFields(DataInput in) throws IOException {
        this.offset = in.readLong();
        this.fileName = Text.readString(in);
    }

    public void write(DataOutput out) throws IOException {
        out.writeLong(offset);
        Text.writeString(out, fileName);
    }

    public int compareTo(FileLineWritable that) {
        int cmp = this.fileName.compareTo(that.fileName);
        if (cmp != 0)
            return cmp;
        return (int) Math.signum((double) (this.offset - that.offset));
    }

    @Override
    public int hashCode() {
        final int prime = 31;
        int result = 1;
        result = prime * result
                + ((fileName == null) ? 0 : fileName.hashCode());
        result = prime * result + (int) (offset ^ (offset >>> 32));
        return result;
    }

    @Override
    public boolean equals(Object obj) {
        if (this == obj)
            return true;
        if (obj == null)
            return false;
        if (getClass() != obj.getClass())
            return false;
        FileLineWritable other = (FileLineWritable) obj;
        if (fileName == null) {
            if (other.fileName != null)
                return false;
        } else if (!fileName.equals(other.fileName))
            return false;
        if (offset != other.offset)
            return false;
        return true;
    }
}

CFInputFormat.java
package com.srini.inputformat;

import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.CombineFileRecordReader;
import org.apache.hadoop.mapreduce.lib.input.CombineFileSplit;



public class CFInputFormat extends
        CombineFileInputFormat<FileLineWritable, Text> {
    public CFInputFormat() {
        super();
        setMaxSplitSize(67108864);
    }

    public RecordReader<FileLineWritable, Text> createRecordReader(
            InputSplit split, TaskAttemptContext context) throws IOException {
        return new CombineFileRecordReader<FileLineWritable, Text>(
                (CombineFileSplit) split, context, CFRecordReader.class);
    }

    @Override
    protected boolean isSplitable(JobContext context, Path file) {
        return false;
    }
}






CFRecordReader.java
package com.srini.inputformat;

import java.io.IOException;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.CombineFileSplit;
import org.apache.hadoop.util.LineReader;

public class CFRecordReader extends RecordReader<FileLineWritable, Text> {
    private long startOffset;
    private long end;
    private long pos;
    private FileSystem fs;
    private Path path;
    private FileLineWritable key;
    private Text value;

    private FSDataInputStream fileIn;
    private LineReader reader;

    public CFRecordReader(CombineFileSplit split, TaskAttemptContext context,
            Integer index) throws IOException {
        this.path = split.getPath(index);
        fs = this.path.getFileSystem(context.getConfiguration());
        this.startOffset = split.getOffset(index);
        this.end = startOffset + split.getLength(index);

        fileIn = fs.open(path);
        reader = new LineReader(fileIn);
        this.pos = startOffset;
    }

    @Override
    public void initialize(InputSplit arg0, TaskAttemptContext arg1)
            throws IOException, InterruptedException {
    }

    @Override
    public void close() throws IOException {
    }

    @Override
    public float getProgress() throws IOException {
        if (startOffset == end) {
            return 0;
        }
        return Math
                .min(1.0f, (pos - startOffset) / (float) (end - startOffset));
    }

    @Override
    public FileLineWritable getCurrentKey() throws IOException,
            InterruptedException {
        return key;
    }

    @Override
    public Text getCurrentValue() throws IOException, InterruptedException {
        return value;
    }

    @Override
    public boolean nextKeyValue() throws IOException {
        if (key == null) {
            key = new FileLineWritable();
            key.fileName = path.getName();
        }
        key.offset = pos;
        if (value == null) {
            value = new Text();
        }
        int newSize = 0;
        if (pos < end) {
            newSize = reader.readLine(value);
            pos += newSize;
        }
        if (newSize == 0) {
            key = null;
            value = null;
            return false;
        } else {
            return true;
        }
    }
}




CountPartioner.java
package com.srini.inputformat;

import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

public class CountPartioner extends Partitioner<Text, IntWritable> {

    @Override
    public int getPartition(Text key, IntWritable value, int numReduceTasks) {
        String keys = key.toString();
        Pattern regex = Pattern.compile("[$&+,:;=?@#|]");
        Matcher matcher = regex.matcher(keys);
        if (matcher.find()) {
            return 0;
        } else {
            return 1 % numReduceTasks;
        }
    }

}

CfWordCount.java
package com.srini.inputformat;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
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;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class CfWordCount extends Configured implements Tool {
    public static class SimpleMap extends
            Mapper<FileLineWritable, Text, Text, IntWritable> {
        public Text word = new Text();
        public IntWritable val = new IntWritable(1);

        public void map(FileLineWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            String string = value.toString();
            StringTokenizer tokenizer = new StringTokenizer(string);
            while (tokenizer.hasMoreElements()) {
                word.set(tokenizer.nextToken());
                context.write(word, val);
            }
        }
    }

    public static class SimpleReduce extends
            Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();

        public void reduce(Text key, Iterable<IntWritable> values,
                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 {
        int exit = ToolRunner.run(new Configuration(), new CfWordCount(), args);
        System.exit(exit);
    }

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.addResource(new Path("/usr/hadoop/conf/core-site.xml"));
        conf.addResource(new Path("/usr/hadoop/conf/mapred-site.xml"));
        conf.addResource(new Path("/usr/hadoop/conf/hdfs-site.xml"));
        FileSystem fs = FileSystem.get(conf);
        fs.delete(new Path("/user/out1"), true);
        Job job = new Job(conf, this.getClass().getName());
        job.setJarByClass(this.getClass());
        job.setJobName(this.getClass().getName());
        job.setInputFormatClass(CFInputFormat.class);
        FileInputFormat.addInputPath(job, new Path("/user/root"));
        job.setMapperClass(SimpleMap.class);
        job.setCombinerClass(SimpleReduce.class);
        job.setPartitionerClass(CountPartioner.class);
        job.setNumReduceTasks(2);
        job.setReducerClass(SimpleReduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        FileOutputFormat.setOutputPath(job, new Path("/user/out1"));
        return job.waitForCompletion(true) ? 0 : 1;
    }
}

MultiWordCount.java
package com.srini.inputformat;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
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.MultipleInputs;
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 MultiWordCount extends Configured implements Tool {
    public static class SimpleMap extends
            Mapper<FileLineWritable, Text, Text, IntWritable> {
        public Text word = new Text();
        public IntWritable val = new IntWritable(1);

        public void map(FileLineWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            String string = value.toString();
            StringTokenizer tokenizer = new StringTokenizer(string);
            while (tokenizer.hasMoreElements()) {
                word.set(tokenizer.nextToken());
                context.write(word, val);
            }
        }
    }

    public static class SimpleReduce extends
            Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();

        public void reduce(Text key, Iterable<IntWritable> values,
                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 {
        int exit = ToolRunner.run(new Configuration(), new MultiWordCount(),
                args);
        System.exit(exit);
    }

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.addResource(new Path("/usr/hadoop/conf/core-site.xml"));
        conf.addResource(new Path("/usr/hadoop/conf/mapred-site.xml"));
        conf.addResource(new Path("/usr/hadoop/conf/hdfs-site.xml"));
        FileSystem fs = FileSystem.get(conf);
        fs.delete(new Path("/user/out1"), true);
        Job job = new Job(conf, this.getClass().getName());
        job.setJarByClass(this.getClass());
        job.setJobName(this.getClass().getName());
        MultipleInputs.addInputPath(job, new Path("/user/root"),
                CFInputFormat.class, SimpleMap.class);
        MultipleInputs.addInputPath(job, new Path("/user/out"),
                CFInputFormat.class, SimpleMap.class);
        job.setCombinerClass(SimpleReduce.class);
        job.setPartitionerClass(CountPartioner.class);
        job.setNumReduceTasks(2);
        job.setReducerClass(SimpleReduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        FileOutputFormat.setOutputPath(job, new Path("/user/out1"));
        return job.waitForCompletion(true) ? 0 : 1;
    }
}