贴一段java读取hdfs 解压gz zip tar.gz保存到hdfs的代码

时间：2018-08-10 11:37:53 阅读：249 评论：0 收藏：0 [点我收藏+]

标签：org ado arc archive oop open gzip .com linked

package main.java;

import java.io.*;
import java.util.LinkedList;
import java.util.List;
import java.util.zip.*;

import org.apache.commons.compress.archivers.ArchiveException;

import org.apache.commons.compress.archivers.ArchiveInputStream;

import org.apache.commons.compress.archivers.ArchiveStreamFactory;

import org.apache.commons.compress.archivers.tar.TarArchiveEntry;

import java.io.IOException;
import java.net.URI;

import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;;

/**

* 解压tar.gz zip gz文件包这里的数据源和输出目录都为HDFS

*

*/

public class GZipHdfs {

private BufferedOutputStream bufferedOutputStream;
String zipfileName = null;

public GZipHdfs(String fileName) {

this.zipfileName = fileName;

}

/*

* 执行入口,rarFileName为需要解压的文件路径(具体到文件),destDir为解压目标路径路径为HDFS

*/

public List<String> unTargzFile(String rarFileName, String destDir) throws IOException {

GZipHdfs GZipHdfs = new GZipHdfs(rarFileName);
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(URI.create(destDir), conf);
boolean result = fs.isDirectory(new Path(destDir));
if (!result) {
fs.mkdirs(new Path(destDir));
}
String outputDirectory = destDir;

List<String> r = GZipHdfs.defUnTargzFile(outputDirectory, fs);
fs.close();
return r;

}

public List<String> defUnTargzFile(String outputDirectory, FileSystem fs) {

FileInputStream fis = null;
ArchiveInputStream in = null;
BufferedInputStream bufferedInputStream = null;
List<String> tarList = new LinkedList<String>();

try {

FSDataInputStream hdfsInputStream = fs.open(new Path(zipfileName));

GZIPInputStream is = new GZIPInputStream(new BufferedInputStream(
hdfsInputStream));

in = new ArchiveStreamFactory().createArchiveInputStream("tar", is);

bufferedInputStream = new BufferedInputStream(in);

TarArchiveEntry entry = (TarArchiveEntry) in.getNextEntry();

while (entry != null) {

String name = entry.getName();

String[] names = name.split("/");

String fileName = outputDirectory;

for (int i = 0; i < names.length; i++) {

String str = names[i];

fileName = fileName + "/" + str;

}

FSDataOutputStream hdfsOutStream = fs.create(new Path(fileName));

bufferedOutputStream = new BufferedOutputStream(
hdfsOutStream);

int b;

while ((b = bufferedInputStream.read()) != -1) {

bufferedOutputStream.write(b);

}

bufferedOutputStream.flush();

bufferedOutputStream.close();

entry = (TarArchiveEntry) in.getNextEntry();

tarList.add(name);

}

} catch (FileNotFoundException e) {

e.printStackTrace();

} catch (IOException e) {

e.printStackTrace();

} catch (ArchiveException e) {

e.printStackTrace();

} finally {

try {

if (bufferedInputStream != null) {

bufferedInputStream.close();

}

} catch (IOException e) {

e.printStackTrace();

}

}
return tarList;

}
/*

* 执行入口,rarFileName为需要解压的文件路径(具体到文件),destDir为解压目标路径路径为HDFS

*/

public List<String> unZipFile(String rarFileName, String destDir) throws IOException {

GZipHdfs GZipHdfs = new GZipHdfs(rarFileName);
Configuration conf = new Configuration();

FileSystem fs = FileSystem.get(URI.create(destDir), conf);
boolean result = fs.isDirectory(new Path(destDir));
if (!result) {
fs.mkdirs(new Path(destDir));
}
String outputDirectory = destDir;

List<String> r = GZipHdfs.defUnZipFile(outputDirectory, fs);
fs.close();
return r;

}

public List<String> defUnZipFile(String outputDirectory, FileSystem fs) {

FileInputStream fis = null;
ArchiveInputStream in = null;
BufferedInputStream bufferedInputStream = null;
List<String> zipList = new LinkedList<String>();

try {

FSDataInputStream hdfsInputStream = fs.open(new Path(zipfileName));
ZipInputStream is = new ZipInputStream(new BufferedInputStream(
hdfsInputStream));

bufferedInputStream = new BufferedInputStream(is);
ZipEntry entry =is.getNextEntry();

while (entry != null) {

String name = entry.getName();

String[] names = name.split("/");

String fileName = outputDirectory;

for (int i = 0; i < names.length; i++) {

String str = names[i];

fileName = fileName + "/" + str;

}

FSDataOutputStream hdfsOutStream = fs.create(new Path(fileName));

bufferedOutputStream = new BufferedOutputStream(
hdfsOutStream);

int b;

while ((b = bufferedInputStream.read()) != -1) {

bufferedOutputStream.write(b);

}

bufferedOutputStream.flush();

bufferedOutputStream.close();

entry = (ZipEntry) is.getNextEntry();

zipList.add(name);

}

} catch (FileNotFoundException e) {

e.printStackTrace();

} catch (IOException e) {

e.printStackTrace();

} finally {

try {

if (bufferedInputStream != null) {

bufferedInputStream.close();

}

} catch (IOException e) {

e.printStackTrace();

}

}
return zipList;

}
/*

* 执行入口,rarFileName为需要解压的文件路径(具体到文件),destDir为解压目标路径路径为HDFS

*/

public List<String> unGZipFile(String rarFileName, String destDir) throws IOException {

GZipHdfs GZipHdfs = new GZipHdfs(rarFileName);
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(URI.create(destDir), conf);
boolean result = fs.isDirectory(new Path(destDir));
if (!result) {
fs.mkdirs(new Path(destDir));
}
String outputDirectory = destDir;

List<String> r = GZipHdfs.defUnGZipFile(outputDirectory, fs);
fs.close();
return r;

}

public List<String> defUnGZipFile(String outputDirectory, FileSystem fs) {

FileInputStream fis = null;
ArchiveInputStream in = null;
BufferedInputStream bufferedInputStream = null;
List<String> tarList = new LinkedList<String>();

try {

FSDataInputStream hdfsInputStream = fs.open(new Path(zipfileName));

GzipCompressorInputStream is = new GzipCompressorInputStream(new BufferedInputStream(
hdfsInputStream));
bufferedInputStream = new BufferedInputStream(is);

String[] nameList = zipfileName.split("/");
String name=nameList[nameList.length-1].replace(".gz","");
String fileName = outputDirectory+"/"+name;
FSDataOutputStream hdfsOutStream = fs.create(new Path(fileName));
bufferedOutputStream = new BufferedOutputStream(
hdfsOutStream);
int b;
while ((b = bufferedInputStream.read()) != -1) {
bufferedOutputStream.write(b);

}
bufferedOutputStream.flush();
bufferedOutputStream.close();
tarList.add(name);

} catch (FileNotFoundException e) {

e.printStackTrace();

} catch (IOException e) {

e.printStackTrace();

} finally {

try {

if (bufferedInputStream != null) {

bufferedInputStream.close();

}

} catch (IOException e) {

e.printStackTrace();

}

}
return tarList;

}

}

标签：org ado arc archive oop open gzip .com linked

原文地址：http://blog.51cto.com/qidai/2157056

踩

(0)

评论一句话评论（0）

分享档案

更多>

2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)

周排行