Hadoop:讀取ｈｄｆｓ上ｚｉｐ壓縮包並解壓到ｈｄｆｓ的實現代碼
阿新 • • 發佈：2017-10-13
++ indexof zipfile substr substring merge str return 刪除
背景：
目前工作中遇到一大批的數據，如果不壓縮直接上傳到ｆｔｐ上就會遇到ｆｔｐ空間資源不足問題，沒辦法只能壓縮後上傳，上穿完成後在ｌｉｎｕｘ上下載。但是ｌｉｎｕｘ客戶端的資源只有２０Ｇ左右一個壓縮包解壓後就要占用１６Ｇ左右的空間，因此想在ｌｉｎｕｘ上直接解壓已經太折騰了（因為我們一共需要處理的這樣的壓縮包包含有３０個左右）。
解決方案：
先把ｌｉｎｕｘ上下載到的ｚｉｐ壓縮包上傳到ｈｄｆｓ，等待所有ｚｉｐ壓縮包都上傳完成後，開始使用程序直接在讀取ｈｄｆｓ上的壓縮包文件，直接解壓到ｈｄｆｓ上，之後把解壓後的文件壓縮為ｇｚｉｐ，實現代碼如下（參考：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; publicstatic void 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 需要解壓縮的文件關鍵名 */ public static void 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 = new byte[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 = new byte[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(); } } public static void 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
執行過程中快照：
[[email protected] ~]$ hadoop fs -ls /user/c/df/myzip 17/10/12 20:26:27 INFO hdfs.PeerCache: SocketCache disabled. Found 30 items -rw-r--r--+ 3 c hadoop 2358927121 2017-10-11 21:57 user/c/df/myzip/myzip_0.zip -rw-r--r--+ 3 c hadoop 2361573235 2017-10-11 19:33 user/c/df/myzip/myzip_12.zip -rw-r--r--+ 3 c hadoop 2359169853 2017-10-11 19:34 user/c/df/myzip/myzip_15.zip ... [ｃ@v09823 ~]$ yarn jar My_ConvertHdfsZipFileToGzipFile.jar /user/c/df/myzip user/c/df/mygzip .zip 17/10/13 00:00:36 INFO hdfs.PeerCache: SocketCache disabled. myzip_0.zip myzip_12.zip myzip_15.zip ... [[email protected] ~]$ hadoop fs -ls -h user/c/df/mygzip 17/10/13 00:44:45 INFO hdfs.PeerCache: SocketCache disabled. Found 3 items -rw-r--r--+ 3 c hadoop 14.9 G 2017-10-13 00:15 user/c/df/mygzip/myzip_0 -rw-r--r--+ 3 c hadoop 14.9 G 2017-10-13 00:35 user/c/df/mygzip/myzip_12 -rw-r--r--+ 3 c hadoop 6 G 2017-10-13 00:35 user/c/df/mygzip/myzip_15 ....
Hadoop:讀取ｈｄｆｓ上ｚｉｐ壓縮包並解壓到ｈｄｆｓ的實現代碼