Introduction Of Dumbo

        Dumbo is a python module for written and running the map-reduce program in the hadoop Cluster. it is working only for the unix/linux environment.  it contain the two kind of function, one is mapper and another one is reducer and finally it has one run method. which is calling the mapper function,reducer and combiner function(optional) and it was submit the map-reduce program in the hadoop Cluster.

how to install dumbo?

     On the machine from which you want to run your Dumbo programs, do:
        $ wget -O ez_setup.py http://bit.ly/ezsetup
        $ python ez_setup.py -z dumbo
or, if you already have easy_install installed, simply:
        $ easy_install -z dumbo

simple wordcount program in dumbo?.

'''
Created on 17-Aug-2014

@author: srinivasan
'''
def mapper(key, value):
    for word in value.split():
        yield word, 1

def reducer(key, values):
    yield key, sum(values)

if __name__ == "__main__":
    import dumbo
    dumbo.run(mapper, reducer)  

How to submit the program in the hadoop Cluster?

dumbo start WordCount.py -hadoop /usr/hadoop -input /user/root/capacity-scheduler.xml -output /user/hadoopresult

note: it will store the output in the sequence file format.

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

ListMultiMapWrittable in Hadoop

ListMultiMapWritable.java
package com.srini.writable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.hadoop.io.AbstractMapWritable;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.util.ReflectionUtils;

import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ListMultimap;
import com.google.common.collect.Multimap;
import com.google.common.collect.Multiset;

public class ListMultiMapWritable extends AbstractMapWritable implements
        ListMultimap<Writable, Writable> {
    private ListMultimap<Writable, Writable> map;

    public ListMultiMapWritable() {
        super();
        this.map = ArrayListMultimap.create();
    }

    public ListMultiMapWritable(ListMultimap<Writable, Writable> map) {
        super();
        this.map = map;
    }

    @Override
    public void clear() {
        this.map.clear();

    }

    @Override
    public boolean containsEntry(Object key, Object value) {
        return this.containsEntry(key, value);
    }

    @Override
    public boolean containsKey(Object key) {
        return this.map.containsKey(key);
    }

    @Override
    public boolean containsValue(Object value) {
        return this.map.containsKey(value);
    }

    @Override
    public Collection<Entry<Writable, Writable>> entries() {
        return this.map.entries();
    }

    @Override
    public boolean isEmpty() {
        return this.map.isEmpty();
    }

    @Override
    public Set<Writable> keySet() {
        return this.map.keySet();
    }

    @Override
    public Multiset<Writable> keys() {
        return this.map.keys();
    }

    @Override
    public boolean put(Writable key, Writable value) {
        return this.map.put(key, value);
    }

    @Override
    public boolean putAll(
            Multimap<? extends Writable, ? extends Writable> values) {
        return this.map.putAll(values);
    }

    @Override
    public boolean putAll(Writable key, Iterable<? extends Writable> value) {
        return this.map.putAll(key, value);
    }

    @Override
    public boolean remove(Object key, Object value) {
        return this.map.remove(key, value);
    }

    @Override
    public int size() {
        return this.map.size();
    }

    @Override
    public Collection<Writable> values() {
        return this.map.values();
    }

    @Override
    public Map<Writable, Collection<Writable>> asMap() {
        return this.map.asMap();
    }

    @Override
    public List<Writable> get(Writable key) {
        return this.map.get(key);
    }

    @Override
    public List<Writable> removeAll(Object key) {
        return this.map.removeAll(key);
    }

    @Override
    public List<Writable> replaceValues(Writable key,
            Iterable<? extends Writable> value) {
        return this.map.replaceValues(key, value);
    }

    @SuppressWarnings("unchecked")
    @Override
    public void readFields(DataInput in) throws IOException {
        super.readFields(in);
        this.map.clear();
        int entries = in.readInt();
        for (int i = 0; i < entries; i++) {
            Writable key = (Writable) ReflectionUtils.newInstance(
                    getClass(in.readByte()), getConf());
            key.readFields(in);
            Writable value = (Writable) ReflectionUtils.newInstance(
                    getClass(in.readByte()), getConf());
            value.readFields(in);
            this.map.put(key, value);
        }

    }

    @Override
    public void write(DataOutput out) throws IOException {
        super.write(out);
        out.writeInt(this.map.size());
        Collection<Entry<Writable, Writable>> entries = this.map.entries();
        for (Entry<Writable, Writable> entry : entries) {
            out.writeByte(getId(entry.getKey().getClass()));
            entry.getKey().write(out);
            out.writeByte(getId(entry.getValue().getClass()));
            entry.getValue().write(out);
        }

    }

}

Hadoop Compression code

Hi Guys,
             After long day's I started to write my blog-spot. today i am going to give some Hadoop Compression code stuff. i think it may help ful for your's projects.
if it is help ful for your project, please send the feedback/ write the comments.
The below codes are deal with reading and writing the file from hdfs using different hadoop compression techniques. 

CompressionOperation.java
package com.srini.compression;

import java.io.File;
import java.io.InputStream;
import java.util.List;

import org.apache.hadoop.fs.Path;

public interface CompressionOperation {
    public List<String> readFile(String namenode_address, Path inpath);

    public void WriteFile(String namenode_address, File inputFile, Path outpath);

    public void WriteFile(String namenode_address, File[] inputFile,
            Path outpath);

    public void WriteFile(String namenode_address, InputStream inputStream,
            Path outpath);

    public void WriteFile(String namenode_address, File[] inputFile,
            Path[] outpath);
}


 DefaultCompression.java
 package com.srini.compression;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.List;

import org.apache.commons.io.FilenameUtils;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.CompressionInputStream;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.io.compress.DefaultCodec;

public class DefaultCompression implements CompressionOperation {

    private CompressionInputStream stream;
    private List<String> content;
    private static Configuration conf = new Configuration();
    private static FileInputStream inputStream;
    private static FSDataOutputStream outputStream;
    private static FileSystem fs;
    private static final String COMPRESSION_EXTENSION = "deflate";

    @Override
    public List<String> readFile(String namenode_address, Path inpath) {
        conf.set("fs.default.name", namenode_address);
        try {
            fs = FileSystem.get(conf);
            if (fs.exists(inpath)) {
                FSDataInputStream inputStream = fs.open(inpath);
                CompressionCodecFactory factory = new CompressionCodecFactory(
                        conf);
                CompressionCodec codec = factory
                        .getCodecByClassName(DefaultCodec.class.getName());
                stream = codec.createInputStream(inputStream,
                        codec.createDecompressor());
                content = IOUtils.readLines(stream, "UTF-8");
                fs.close();
                IOUtils.closeQuietly(inputStream);
                IOUtils.closeQuietly(stream);
            } else {
                System.out.println("The file " + inpath
                        + " is not avilable in the hdfs.");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return content;
    }

    @Override
    public void WriteFile(String namenode_address, File inputFile, Path outpath) {
        conf.set("fs.default.name", namenode_address);
        try {
            inputStream = new FileInputStream(inputFile);
            fs = FileSystem.get(conf);
            if (!fs.exists(outpath)) {
                outputStream = fs.create(outpath);
                CompressionCodecFactory factory = new CompressionCodecFactory(
                        conf);
                CompressionCodec codec = factory
                        .getCodecByClassName(DefaultCodec.class.getName());
                CompressionOutputStream stream = codec.createOutputStream(
                        outputStream, codec.createCompressor());
                List<String> lines = IOUtils.readLines(inputStream, "UTF-8");
                for (String string : lines) {
                    IOUtils.write(string, stream, "UTF-8");
                }
                IOUtils.closeQuietly(inputStream);
                IOUtils.closeQuietly(stream);
                IOUtils.closeQuietly(outputStream);
                fs.close();
            } else {
                System.out
                        .println("The ouput--"
                                + outpath
                                + "-- is already exsist in the hdfs. please choose the diffrent path");
            }

        } catch (IOException e) {
            IOUtils.closeQuietly(inputStream);
            IOUtils.closeQuietly(stream);
            IOUtils.closeQuietly(outputStream);
            e.printStackTrace();
        }

    }

    @Override
    public void WriteFile(String namenode_address, File[] inputFile,
            Path outpath) {
        conf.set("fs.default.name", namenode_address);
        for (File file : inputFile) {
            try {
                inputStream = new FileInputStream(file);
                fs = FileSystem.get(conf);
                FileStatus status = fs.getFileStatus(outpath);
                if (status.isDir()) {
                    outputStream = fs.create(new Path(outpath.toString()
                            + Path.SEPARATOR
                            + FilenameUtils.getBaseName(file.getName()) + "."
                            + COMPRESSION_EXTENSION));
                    CompressionCodecFactory factory = new CompressionCodecFactory(
                            conf);
                    CompressionCodec codec = factory
                            .getCodecByClassName(DefaultCodec.class.getName());
                    CompressionOutputStream stream = codec.createOutputStream(
                            outputStream, codec.createCompressor());
                    List<String> lines = IOUtils
                            .readLines(inputStream, "UTF-8");
                    for (String string : lines) {
                        IOUtils.write(string, stream, "UTF-8");
                    }
                    IOUtils.closeQuietly(inputStream);
                    IOUtils.closeQuietly(stream);
                    IOUtils.closeQuietly(outputStream);
                } else {
                    System.out.println("The ouput--" + outpath
                            + "-- is not a directory");
                }
            } catch (IOException e) {
                IOUtils.closeQuietly(inputStream);
                IOUtils.closeQuietly(stream);
                IOUtils.closeQuietly(outputStream);
                e.printStackTrace();
            }
        }
        try {
            fs.close();
        } catch (IOException e) {
            IOUtils.closeQuietly(inputStream);
            IOUtils.closeQuietly(stream);
            IOUtils.closeQuietly(outputStream);
            e.printStackTrace();
        }
    }

    @Override
    public void WriteFile(String namenode_address, InputStream inputStream,
            Path outpath) {
        conf.set("fs.default.name", namenode_address);
        try {
            fs = FileSystem.get(conf);
            outputStream = fs.create(outpath);
            CompressionCodecFactory factory = new CompressionCodecFactory(conf);
            CompressionCodec codec = factory
                    .getCodecByClassName(DefaultCodec.class.getName());
            CompressionOutputStream stream = codec.createOutputStream(
                    outputStream, codec.createCompressor());
            List<String> lines = IOUtils.readLines(inputStream, "UTF-8");
            for (String string : lines) {
                IOUtils.write(string, stream, "UTF-8");
            }
            IOUtils.closeQuietly(inputStream);
            IOUtils.closeQuietly(stream);
            IOUtils.closeQuietly(outputStream);
        } catch (IOException e) {
            IOUtils.closeQuietly(inputStream);
            IOUtils.closeQuietly(stream);
            IOUtils.closeQuietly(outputStream);
            e.printStackTrace();
        }
    }

    @Override
    public void WriteFile(String namenode_address, File[] inputFile,
            Path[] outpath) {
        for (int i = 0; i <= File.listRoots().length; i++) {
            WriteFile(namenode_address, inputFile[i], outpath[i]);
        }
    }
}

Bzip2Compression.java
package com.srini.compression;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.List;

import org.apache.commons.io.FilenameUtils;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.compress.BZip2Codec;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.CompressionInputStream;
import org.apache.hadoop.io.compress.CompressionOutputStream;

public class Bzip2Compression implements CompressionOperation {

    private CompressionInputStream stream;
    private List<String> content;
    private static Configuration conf = new Configuration();
    private static FileInputStream inputStream;
    private static FSDataOutputStream outputStream;
    private static FileSystem fs;
    private static final String COMPRESSION_EXTENSION = "bzip2";

    @Override
    public List<String> readFile(String namenode_address, Path inpath) {
        conf.set("fs.default.name", namenode_address);
        try {
            fs = FileSystem.get(conf);
            if (fs.exists(inpath)) {
                FSDataInputStream inputStream = fs.open(inpath);
                CompressionCodecFactory factory = new CompressionCodecFactory(
                        conf);
                CompressionCodec codec = factory
                        .getCodecByClassName(BZip2Codec.class.getName());
                stream = codec.createInputStream(inputStream,
                        codec.createDecompressor());
                content = IOUtils.readLines(stream, "UTF-8");
                fs.close();
                IOUtils.closeQuietly(inputStream);
                IOUtils.closeQuietly(stream);
            } else {
                System.out.println("The file " + inpath
                        + " is not avilable in the hdfs.");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return content;
    }

    @Override
    public void WriteFile(String namenode_address, File inputFile, Path outpath) {
        conf.set("fs.default.name", namenode_address);
        try {
            inputStream = new FileInputStream(inputFile);
            fs = FileSystem.get(conf);
            if (!fs.exists(outpath)) {
                outputStream = fs.create(outpath);
                CompressionCodecFactory factory = new CompressionCodecFactory(
                        conf);
                CompressionCodec codec = factory
                        .getCodecByClassName(BZip2Codec.class.getName());
                CompressionOutputStream stream = codec.createOutputStream(
                        outputStream, codec.createCompressor());
                List<String> lines = IOUtils.readLines(inputStream, "UTF-8");
                for (String string : lines) {
                    IOUtils.write(string, stream, "UTF-8");
                }
                IOUtils.closeQuietly(inputStream);
                IOUtils.closeQuietly(stream);
                IOUtils.closeQuietly(outputStream);
                fs.close();
            } else {
                System.out
                        .println("The ouput--"
                                + outpath
                                + "-- is already exsist in the hdfs. please choose the diffrent path");
            }

        } catch (IOException e) {
            IOUtils.closeQuietly(inputStream);
            IOUtils.closeQuietly(stream);
            IOUtils.closeQuietly(outputStream);
            e.printStackTrace();
        }

    }

    @Override
    public void WriteFile(String namenode_address, File[] inputFile,
            Path outpath) {
        conf.set("fs.default.name", namenode_address);
        for (File file : inputFile) {
            try {
                inputStream = new FileInputStream(file);
                fs = FileSystem.get(conf);
                FileStatus status = fs.getFileStatus(outpath);
                if (status.isDir()) {
                    outputStream = fs.create(new Path(outpath.toString()
                            + Path.SEPARATOR
                            + FilenameUtils.getBaseName(file.getName()) + "."
                            + COMPRESSION_EXTENSION));
                    CompressionCodecFactory factory = new CompressionCodecFactory(
                            conf);
                    CompressionCodec codec = factory
                            .getCodecByClassName(BZip2Codec.class.getName());
                    CompressionOutputStream stream = codec.createOutputStream(
                            outputStream, codec.createCompressor());
                    List<String> lines = IOUtils
                            .readLines(inputStream, "UTF-8");
                    for (String string : lines) {
                        IOUtils.write(string, stream, "UTF-8");
                    }
                    IOUtils.closeQuietly(inputStream);
                    IOUtils.closeQuietly(stream);
                    IOUtils.closeQuietly(outputStream);
                } else {
                    System.out.println("The ouput--" + outpath
                            + "-- is not a directory");
                }
            } catch (IOException e) {
                IOUtils.closeQuietly(inputStream);
                IOUtils.closeQuietly(stream);
                IOUtils.closeQuietly(outputStream);
                e.printStackTrace();
            }
        }
        try {
            fs.close();
        } catch (IOException e) {
            IOUtils.closeQuietly(inputStream);
            IOUtils.closeQuietly(stream);
            IOUtils.closeQuietly(outputStream);
            e.printStackTrace();
        }
    }

    @Override
    public void WriteFile(String namenode_address, InputStream inputStream,
            Path outpath) {
        conf.set("fs.default.name", namenode_address);
        try {
            fs = FileSystem.get(conf);
            outputStream = fs.create(outpath);
            CompressionCodecFactory factory = new CompressionCodecFactory(conf);
            CompressionCodec codec = factory
                    .getCodecByClassName(BZip2Codec.class.getName());
            CompressionOutputStream stream = codec.createOutputStream(
                    outputStream, codec.createCompressor());
            List<String> lines = IOUtils.readLines(inputStream, "UTF-8");
            for (String string : lines) {
                IOUtils.write(string, stream, "UTF-8");
            }
            IOUtils.closeQuietly(inputStream);
            IOUtils.closeQuietly(stream);
            IOUtils.closeQuietly(outputStream);
        } catch (IOException e) {
            IOUtils.closeQuietly(inputStream);
            IOUtils.closeQuietly(stream);
            IOUtils.closeQuietly(outputStream);
            e.printStackTrace();
        }
    }

    @Override
    public void WriteFile(String namenode_address, File[] inputFile,
            Path[] outpath) {
        for (int i = 0; i <= File.listRoots().length; i++) {
            WriteFile(namenode_address, inputFile[i], outpath[i]);
        }
    }
}
 GzipCompression.java
package com.srini.compression;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.List;

import org.apache.commons.io.FilenameUtils;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.CompressionInputStream;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.io.compress.GzipCodec;

public class GzipCompression implements CompressionOperation {

    private CompressionInputStream stream;
    private List<String> content;
    private static Configuration conf = new Configuration();
    private static FileInputStream inputStream;
    private static FSDataOutputStream outputStream;
    private static FileSystem fs;
    private static final String COMPRESSION_EXTENSION = "gz";

    @Override
    public List<String> readFile(String namenode_address, Path inpath) {
        conf.set("fs.default.name", namenode_address);
        try {
            fs = FileSystem.get(conf);
            if (fs.exists(inpath)) {
                FSDataInputStream inputStream = fs.open(inpath);
                CompressionCodecFactory factory = new CompressionCodecFactory(
                        conf);
                CompressionCodec codec = factory
                        .getCodecByClassName(GzipCodec.class.getName());
                stream = codec.createInputStream(inputStream,
                        codec.createDecompressor());
                content = IOUtils.readLines(stream, "UTF-8");
                fs.close();
                IOUtils.closeQuietly(inputStream);
                IOUtils.closeQuietly(stream);
            } else {
                System.out.println("The file " + inpath
                        + " is not avilable in the hdfs.");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return content;
    }

    @Override
    public void WriteFile(String namenode_address, File inputFile, Path outpath) {
        conf.set("fs.default.name", namenode_address);
        try {
            inputStream = new FileInputStream(inputFile);
            fs = FileSystem.get(conf);
            if (!fs.exists(outpath)) {
                outputStream = fs.create(outpath);
                CompressionCodecFactory factory = new CompressionCodecFactory(
                        conf);
                CompressionCodec codec = factory
                        .getCodecByClassName(GzipCodec.class.getName());
                CompressionOutputStream stream = codec.createOutputStream(
                        outputStream, codec.createCompressor());
                List<String> lines = IOUtils.readLines(inputStream, "UTF-8");
                for (String string : lines) {
                    IOUtils.write(string, stream, "UTF-8");
                }
                IOUtils.closeQuietly(inputStream);
                IOUtils.closeQuietly(stream);
                IOUtils.closeQuietly(outputStream);
                fs.close();
            } else {
                System.out
                        .println("The ouput--"
                                + outpath
                                + "-- is already exsist in the hdfs. please choose the diffrent path");
            }

        } catch (IOException e) {
            IOUtils.closeQuietly(inputStream);
            IOUtils.closeQuietly(stream);
            IOUtils.closeQuietly(outputStream);
            e.printStackTrace();
        }

    }

    @Override
    public void WriteFile(String namenode_address, File[] inputFile,
            Path outpath) {
        conf.set("fs.default.name", namenode_address);
        for (File file : inputFile) {
            try {
                inputStream = new FileInputStream(file);
                fs = FileSystem.get(conf);
                FileStatus status = fs.getFileStatus(outpath);
                if (status.isDir()) {
                    outputStream = fs.create(new Path(outpath.toString()
                            + Path.SEPARATOR
                            + FilenameUtils.getBaseName(file.getName()) + "."
                            + COMPRESSION_EXTENSION));
                    CompressionCodecFactory factory = new CompressionCodecFactory(
                            conf);
                    CompressionCodec codec = factory
                            .getCodecByClassName(GzipCodec.class.getName());
                    CompressionOutputStream stream = codec.createOutputStream(
                            outputStream, codec.createCompressor());
                    List<String> lines = IOUtils
                            .readLines(inputStream, "UTF-8");
                    for (String string : lines) {
                        IOUtils.write(string, stream, "UTF-8");
                    }
                    IOUtils.closeQuietly(inputStream);
                    IOUtils.closeQuietly(stream);
                    IOUtils.closeQuietly(outputStream);
                } else {
                    System.out.println("The ouput--" + outpath
                            + "-- is not a directory");
                }
            } catch (IOException e) {
                IOUtils.closeQuietly(inputStream);
                IOUtils.closeQuietly(stream);
                IOUtils.closeQuietly(outputStream);
                e.printStackTrace();
            }
        }
        try {
            fs.close();
        } catch (IOException e) {
            IOUtils.closeQuietly(inputStream);
            IOUtils.closeQuietly(stream);
            IOUtils.closeQuietly(outputStream);
            e.printStackTrace();
        }
    }

    @Override
    public void WriteFile(String namenode_address, InputStream inputStream,
            Path outpath) {
        conf.set("fs.default.name", namenode_address);
        try {
            fs = FileSystem.get(conf);
            outputStream = fs.create(outpath);
            CompressionCodecFactory factory = new CompressionCodecFactory(conf);
            CompressionCodec codec = factory
                    .getCodecByClassName(GzipCodec.class.getName());
            CompressionOutputStream stream = codec.createOutputStream(
                    outputStream, codec.createCompressor());
            List<String> lines = IOUtils.readLines(inputStream, "UTF-8");
            for (String string : lines) {
                IOUtils.write(string, stream, "UTF-8");
            }
            IOUtils.closeQuietly(inputStream);
            IOUtils.closeQuietly(stream);
            IOUtils.closeQuietly(outputStream);
        } catch (IOException e) {
            IOUtils.closeQuietly(inputStream);
            IOUtils.closeQuietly(stream);
            IOUtils.closeQuietly(outputStream);
            e.printStackTrace();
        }
    }

    @Override
    public void WriteFile(String namenode_address, File[] inputFile,
            Path[] outpath) {
        for (int i = 0; i <= File.listRoots().length; i++) {
            WriteFile(namenode_address, inputFile[i], outpath[i]);
        }
    }
}
 
SnappyCompression.java
package com.srini.compression;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.List;

import org.apache.commons.io.FilenameUtils;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.CompressionInputStream;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.io.compress.SnappyCodec;

public class SnappyCompression implements CompressionOperation {

    private CompressionInputStream stream;
    private List<String> content;
    private static Configuration conf = new Configuration();
    private static FileInputStream inputStream;
    private static FSDataOutputStream outputStream;
    private static FileSystem fs;
    private static final String COMPRESSION_EXTENSION = "snappy";

    @Override
    public List<String> readFile(String namenode_address, Path inpath) {
        conf.set("fs.default.name", namenode_address);
        try {
            fs = FileSystem.get(conf);
            if (fs.exists(inpath)) {
                FSDataInputStream inputStream = fs.open(inpath);
                CompressionCodecFactory factory = new CompressionCodecFactory(
                        conf);
                CompressionCodec codec = factory
                        .getCodecByClassName(SnappyCodec.class.getName());
                stream = codec.createInputStream(inputStream,
                        codec.createDecompressor());
                content = IOUtils.readLines(stream, "UTF-8");
                fs.close();
                IOUtils.closeQuietly(inputStream);
                IOUtils.closeQuietly(stream);
            } else {
                System.out.println("The file " + inpath
                        + " is not avilable in the hdfs.");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return content;
    }

    @Override
    public void WriteFile(String namenode_address, File inputFile, Path outpath) {
        conf.set("fs.default.name", namenode_address);
        try {
            inputStream = new FileInputStream(inputFile);
            fs = FileSystem.get(conf);
            if (!fs.exists(outpath)) {
                outputStream = fs.create(outpath);
                CompressionCodecFactory factory = new CompressionCodecFactory(
                        conf);
                CompressionCodec codec = factory
                        .getCodecByClassName(SnappyCodec.class.getName());
                CompressionOutputStream stream = codec.createOutputStream(
                        outputStream, codec.createCompressor());
                List<String> lines = IOUtils.readLines(inputStream, "UTF-8");
                for (String string : lines) {
                    IOUtils.write(string, stream, "UTF-8");
                }
                IOUtils.closeQuietly(inputStream);
                IOUtils.closeQuietly(stream);
                IOUtils.closeQuietly(outputStream);
                fs.close();
            } else {
                System.out
                        .println("The ouput--"
                                + outpath
                                + "-- is already exsist in the hdfs. please choose the diffrent path");
            }

        } catch (IOException e) {
            IOUtils.closeQuietly(inputStream);
            IOUtils.closeQuietly(stream);
            IOUtils.closeQuietly(outputStream);
            e.printStackTrace();
        }

    }

    @Override
    public void WriteFile(String namenode_address, File[] inputFile,
            Path outpath) {
        conf.set("fs.default.name", namenode_address);
        for (File file : inputFile) {
            try {
                inputStream = new FileInputStream(file);
                fs = FileSystem.get(conf);
                FileStatus status = fs.getFileStatus(outpath);
                if (status.isDir()) {
                    outputStream = fs.create(new Path(outpath.toString()
                            + Path.SEPARATOR
                            + FilenameUtils.getBaseName(file.getName()) + "."
                            + COMPRESSION_EXTENSION));
                    CompressionCodecFactory factory = new CompressionCodecFactory(
                            conf);
                    CompressionCodec codec = factory
                            .getCodecByClassName(SnappyCodec.class.getName());
                    CompressionOutputStream stream = codec.createOutputStream(
                            outputStream, codec.createCompressor());
                    List<String> lines = IOUtils
                            .readLines(inputStream, "UTF-8");
                    for (String string : lines) {
                        IOUtils.write(string, stream, "UTF-8");
                    }
                    IOUtils.closeQuietly(inputStream);
                    IOUtils.closeQuietly(stream);
                    IOUtils.closeQuietly(outputStream);
                } else {
                    System.out.println("The ouput--" + outpath
                            + "-- is not a directory");
                }
            } catch (IOException e) {
                IOUtils.closeQuietly(inputStream);
                IOUtils.closeQuietly(stream);
                IOUtils.closeQuietly(outputStream);
                e.printStackTrace();
            }
        }
        try {
            fs.close();
        } catch (IOException e) {
            IOUtils.closeQuietly(inputStream);
            IOUtils.closeQuietly(stream);
            IOUtils.closeQuietly(outputStream);
            e.printStackTrace();
        }
    }

    @Override
    public void WriteFile(String namenode_address, InputStream inputStream,
            Path outpath) {
        conf.set("fs.default.name", namenode_address);
        try {
            fs = FileSystem.get(conf);
            outputStream = fs.create(outpath);
            CompressionCodecFactory factory = new CompressionCodecFactory(conf);
            CompressionCodec codec = factory
                    .getCodecByClassName(SnappyCodec.class.getName());
            CompressionOutputStream stream = codec.createOutputStream(
                    outputStream, codec.createCompressor());
            List<String> lines = IOUtils.readLines(inputStream, "UTF-8");
            for (String string : lines) {
                IOUtils.write(string, stream, "UTF-8");
            }
            IOUtils.closeQuietly(inputStream);
            IOUtils.closeQuietly(stream);
            IOUtils.closeQuietly(outputStream);
        } catch (IOException e) {
            IOUtils.closeQuietly(inputStream);
            IOUtils.closeQuietly(stream);
            IOUtils.closeQuietly(outputStream);
            e.printStackTrace();
        }
    }

    @Override
    public void WriteFile(String namenode_address, File[] inputFile,
            Path[] outpath) {
        for (int i = 0; i <= File.listRoots().length; i++) {
            WriteFile(namenode_address, inputFile[i], outpath[i]);
        }
    }
}
 
TestClass.java
package com.srini.compression;

import java.util.List;

import org.apache.hadoop.fs.Path;

public class TestClass {
    public static void main(String[] args) {
        CompressionOperation operation = new GzipCompression();
        List<String> line = operation.readFile("hdfs://localhost:8020",
                new Path("/user/hadoop.gz"));
        System.out.println(line);
    }

}