Saturday, 16 August 2014

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

}
 
Posted by at

No comments:

Post a Comment

Subscribe to: Post Comments (Atom)