Hadoop:读取ｈｄｆｓ上ｚｉｐ压缩包并解压到ｈｄｆｓ的实现代码
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了Hadoop:读取ｈｄｆｓ上ｚｉｐ压缩包并解压到ｈｄｆｓ的实现代码，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含6177字，纯文字阅读大概需要9分钟。
内容图文
背景：
目前工作中遇到一大批的数据，如果不压缩直接上传到ｆｔｐ上就会遇到ｆｔｐ空间资源不足问题，没办法只能压缩后上传，上穿完成后在ｌｉｎｕｘ上下载。但是ｌｉｎｕｘ客户端的资源只有２０Ｇ左右一个压缩包解压后就要占用１６Ｇ左右的空间，因此想在ｌｉｎｕｘ上直接解压已经太折腾了（因为我们一共需要处理的这样的压缩包包含有３０个左右）。
解决方案：
先把ｌｉｎｕｘ上下载到的ｚｉｐ压缩包上传到ｈｄｆｓ，等待所有ｚｉｐ压缩包都上传完成后，开始使用程序直接在读取ｈｄｆｓ上的压缩包文件，直接解压到ｈｄｆｓ上，之后把解压后的文件压缩为ｇｚｉｐ，实现代码如下（参考：http://www.cnblogs.com/juefan/articles/2935163.html）：
import
java.io.File;
import
java.io.IOException;
import
java.util.zip.GZIPOutputStream;
import
java.util.zip.ZipEntry;
import
java.util.zip.ZipInputStream;
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.Text;
/**
* Created by Administrator on 12/10/2017.
*/
public
class
ConvertHdfsZipFileToGzipFile {
public
static
boolean isRecur = false;
publicstaticvoid main(String[] args) throws IOException {
if (args.length == 0)
errorMessage("1filesmerge [-r|-R] <hdfsTargetDir> <hdfsFileName>");
if (args[0].matches("^-[rR]$")) {
isRecur = true;
}
if ((isRecur && args.length != 4) || ( !isRecur && args.length != 3)) {
errorMessage("2filesmerge [-r|-R] <hdfsTargetDir> <hdfsFileName>");
}
Configuration conf = new Configuration();
FileSystem hdfs = FileSystem.get(conf);
Path inputDir;
Path hdfsFile;
Text pcgroupText;
// hadoop jar myjar.jar ConvertHdfsZipFileToGzipFile -r /zip/(待转换文件路径，在HDFS上) /user/j/pconline/(转换完成后的文件存储地址，也在HDFS上) pconline(待转换的文件名包含的字符)if(isRecur){
inputDir = new Path(args[1]);
hdfsFile = new Path(args[2]);
pcgroupText = new Text(args[3]);
}
// hadoop jar myjar.jar ConvertHdfsZipFileToGzipFile /zip/(待转换文件路径，在HDFS上) /user/j/pconline/(转换完成后的文件存储地址，也在HDFS上) pconline(待转换的文件名包含的字符)else{
inputDir = new Path(args[0]);
hdfsFile = new Path(args[1]);
pcgroupText = new Text(args[2]);
}
if (!hdfs.exists(inputDir)) {
errorMessage("3hdfsTargetDir not exist!");
}
if (hdfs.exists(hdfsFile)) {
errorMessage("4hdfsFileName exist!");
}
merge(inputDir, hdfsFile, hdfs, pcgroupText);
System.exit(0);
}
/**
* @author
* @param inputDir zip文件的存储地址
* @param hdfsFile 解压结果的存储地址
* @param hdfs 分布式文件系统数据流
* @param pcgroupText 需要解压缩的文件关键名
*/publicstaticvoid merge(Path inputDir, Path hdfsFile,
FileSystem hdfs, Text pcgroupText) {
try {
//文件系统地址inputDir下的FileStatus
FileStatus[] inputFiles = hdfs.listStatus(inputDir);
for (int i = 0; i < inputFiles.length; i++) {
if (!hdfs.isFile(inputFiles[i].getPath())) {
if (isRecur){
merge(inputFiles[i].getPath(), hdfsFile, hdfs,pcgroupText);
return ;
}
else {
System.out.println(inputFiles[i].getPath().getName()
+ "is not file and not allow recursion, skip!");
continue;
}
}
//判断文件名是否在需要解压缩的关键名内if(inputFiles[i].getPath().getName().contains(pcgroupText.toString()) == true){
//输出待解压的文件名 System.out.println(inputFiles[i].getPath().getName());
//将数据流指向待解压文件
FSDataInputStream in = hdfs.open(inputFiles[i].getPath());
/**
*数据的解压执行过程
*/
ZipInputStream zipInputStream = null;
try{
zipInputStream = new ZipInputStream(in);
ZipEntry entry;
//解压后有多个文件一并解压出来并实现合并
//合并后的地址
FSDataOutputStream mergerout = hdfs.create(new Path(hdfsFile + File.separator +
inputFiles[i].getPath().getName().substring(0, inputFiles[i].getPath().getName().indexOf("."))));
while((entry = zipInputStream.getNextEntry()) != null){
int bygeSize1=2*1024*1024;
byte[] buffer1 = newbyte[bygeSize1];
int nNumber;
while((nNumber = zipInputStream.read(buffer1,0, bygeSize1)) != -1){
mergerout.write(buffer1, 0, nNumber);
}
}
mergerout.flush();
mergerout.close();
zipInputStream.close();
}catch(IOException e){
continue;
}
in.close();
/**
*将解压合并后的数据压缩成gzip格式
*/
GZIPOutputStream gzipOutputStream = null;
try{
FSDataOutputStream outputStream = null;
outputStream = hdfs.create(new Path(hdfsFile + File.separator +
inputFiles[i].getPath().getName().substring(0, inputFiles[i].getPath().getName().indexOf(".")) + ".gz"));
FSDataInputStream inputStream = null;
gzipOutputStream = new GZIPOutputStream(outputStream);
inputStream = hdfs.open(new Path(hdfsFile + File.separator + inputFiles[i].getPath().getName().substring(0, inputFiles[i].getPath().getName().indexOf("."))));
int bygeSize=2*1024*1024;
byte[] buffer = newbyte[bygeSize];
int len;
while((len = inputStream.read(buffer)) > 0){
gzipOutputStream.write(buffer, 0, len);
}
inputStream.close();
gzipOutputStream.finish();
gzipOutputStream.flush();
outputStream.close();
}catch (Exception exception){
exception.printStackTrace();
}
gzipOutputStream.close();
//删除zip文件解压合并后的临时文件
String tempfiles = hdfsFile + File.separator + inputFiles[i].getPath().getName().substring(0, inputFiles[i].getPath().getName().indexOf("."));
try{
if(hdfs.exists(new Path(tempfiles))){
hdfs.delete(new Path(tempfiles), true);
}
}catch(IOException ie){
ie.printStackTrace();
}
}
}
}catch (IOException e) {
e.printStackTrace();
}
}
publicstaticvoid errorMessage(String str) {
System.out.println("Error Message: " + str);
System.exit(1);
}
}
调用：
[ｃ@v09823]# hadoop jar myjar.jar [ConvertHdfsZipFileToGzipFile该ｍａｉｎ的类名根据打包方式决定是否需要] /zip/(待转换文件路径，在HDFS上) /user/j/pconline/(转换完成后的文件存储地址，也在HDFS上) pconline(待转换的文件名包含的字符)
如果要实现递归的话，可以在filesmerge后面加上 -r
执行过程中快照：
[c@v09823 ~]$ hadoop fs -ls /user/c/df/myzip 17/10/1220:26:27 INFO hdfs.PeerCache: SocketCache disabled. Found 30 items -rw-r--r--+ 3 c hadoop 23589271212017-10-1121:57 user/c/df/myzip/myzip_0.zip -rw-r--r--+ 3 c hadoop 23615732352017-10-1119:33 user/c/df/myzip/myzip_12.zip -rw-r--r--+ 3 c hadoop 23591698532017-10-1119:34 user/c/df/myzip/myzip_15.zip ... [ｃ@v09823 ~]$ yarn jar My_ConvertHdfsZipFileToGzipFile.jar /user/c/df/myzip user/c/df/mygzip .zip17/10/1300:00:36 INFO hdfs.PeerCache: SocketCache disabled. myzip_0.zip myzip_12.zip myzip_15.zip ... [catt@vq20skjh01 ~]$ hadoop fs -ls -h user/c/df/mygzip 17/10/1300:44:45 INFO hdfs.PeerCache: SocketCache disabled. Found 3 items -rw-r--r--+ 3 c hadoop 14.9 G 2017-10-1300:15 user/c/df/mygzip/myzip_0 -rw-r--r--+ 3 c hadoop 14.9 G 2017-10-1300:35 user/c/df/mygzip/myzip_12 -rw-r--r--+ 3 c hadoop 6 G 2017-10-1300:35 user/c/df/mygzip/myzip_15 ....
原文：http://www.cnblogs.com/yy3b2007com/p/7659073.html
内容总结
以上是互联网集市为您收集整理的Hadoop:读取ｈｄｆｓ上ｚｉｐ压缩包并解压到ｈｄｆｓ的实现代码全部内容，希望文章能够帮你解决Hadoop:读取ｈｄｆｓ上ｚｉｐ压缩包并解压到ｈｄｆｓ的实现代码所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。
来源：【匿名】