package org.wliu.hdfs_hdp220;

import java.io.IOException;
import java.io.OutputStream;
import java.net.URISyntaxException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
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.CompressionOutputStream;
import org.apache.hadoop.util.ReflectionUtils;

public class HDFSPut {
	
	public static final String FSNAME = "talend";
	
	public static OutputStream getBZip2OutStream(Configuration conf, String hdfsDest) throws IOException, InterruptedException, URISyntaxException {
		FileSystem fs = null;
		fs = FileSystem.get(new java.net.URI(conf.get("fs.default.name")), conf, FSNAME);
		
		Path path_dest = new Path(hdfsDest);
		FSDataOutputStream fsOut = fs.create(path_dest, true);
		CompressionCodec codecOutput = ReflectionUtils.newInstance(BZip2Codec.class, conf);
		CompressionOutputStream cmprOutput = codecOutput.createOutputStream(fsOut);
		return cmprOutput;
	}
	
	public static void closeOutputStream(OutputStream outStream) {
		if (outStream != null) {
			try {
				outStream.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

}

@@ -443,32 +443,80 @@
 onOutputStream;%0A
+import org.apache.hadoop.io.compress.GzipCodec;%0A
 import org.apach
@@ -621,16 +621,390 @@
 end%22;%0A%09%0A
+%09public static OutputStream getFSOutputStream(Configuration conf, String hdfsDest) throws IOException, InterruptedException, URISyntaxException %7B%0A%09%09FileSystem fs = null;%0A%09%09fs = FileSystem.get(new java.net.URI(conf.get(%22fs.default.name%22)), conf, FSNAME);%0A%09%09%0A%09%09Path path_dest = new Path(hdfsDest);%0A%09%09FSDataOutputStream fsOut = fs.create(path_dest, true);%0A%09%09return fsOut;%0A%09%7D%0A%09%0A
 %09public 
@@ -1530,32 +1530,560 @@
 mprOutput;%0A%09%7D%0A%09%0A
+%09public static OutputStream getGZipOutStream(Configuration conf, String hdfsDest) throws IOException, InterruptedException, URISyntaxException %7B%0A%09%09FileSystem fs = null;%0A%09%09fs = FileSystem.get(new java.net.URI(conf.get(%22fs.default.name%22)), conf, FSNAME);%0A%09%09%0A%09%09Path path_dest = new Path(hdfsDest);%0A%09%09FSDataOutputStream fsOut = fs.create(path_dest, true);%0A%09%09GzipCodec codecOutput = new GzipCodec();%0A%09%09codecOutput.setConf(conf);%0A%09%09CompressionOutputStream cmprOutput = codecOutput.createOutputStream(fsOut);%0A%09%09return cmprOutput;%0A%09%7D%0A%09%0A
 %09public static v

add another 2 outputstream support