項目背景
在實際項目中，輸入數據往往是由許多小文件組成，這里的小文件是指小於HDFS系統Block大小的文件（默認128M），早期的版本所定義的小文件是64M，這里的hadoop-2.2.0所定義的小文件是128M。然而每一個存儲在HDFS中的文件、目錄和塊都映射為一個對象，存儲在NameNode服務器內存中，通常占用150個字節。 如果有1千萬個文件，就需要消耗大約3G的內存空間。如果是10億個文件呢，簡直不可想象。所以在項目開始前， 我們要先了解一下 hadoop 處理小文件的各種方案，然后本課程選擇一種適合的方案來解決本項目的小文件問題。Hadoop 自身提供了幾種機制來解決相關的問題，包括HAR， SequeueFile和CombineFileInputFormat。
項目介紹
在本地 D://Code/EclipseCode/mergeSmallFilesTestData目錄下有 2018-03-23 至 2018-03-29 一共7天的數據集，我們需要將這7天的數據集按日期合並為7個大文件上傳至 HDFS。
思路分析
基於項目的需求，我們通過下面幾個步驟完成：
1）首先通過 globStatus()方法過濾掉 svn 格式的文件，獲取 D://Code/EclipseCode/mergeSmallFilesTestData目錄下的其它所有文件路徑。
2）然后循環第一步的所有文件路徑，通過globStatus()方法獲取所有 txt 格式文件路徑。
3）最后通過IOUtils.copyBytes(in, out, 4096, false)方法將數據集合並為7個大文件，並上傳至 HDFS。
程序
在Hadoop項目路徑下新建MergeSmallFilesToHDFS.java：
/** * */ package com.hadoop.train; import java.io.IOException; import java.net.URI; import java.net.URISyntaxException; 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.FileUtil; import org.apache.hadoop.fs.Path; import org.apache.hadoop.fs.PathFilter; import org.apache.hadoop.io.IOUtils; /** * @author Zimo * 合並小文件到HDFS * */
public class MergeSmallFilesToHDFS { private static FileSystem hdfs = null; //定義HDFS上的文件系統對象
private static FileSystem local = null; //定義本地文件系統對象
/** * * @function 過濾 regex 格式的文件 * */
public static class RegexExcludePathFilter implements PathFilter { private final String regex; public RegexExcludePathFilter(String regex) { // TODO Auto-generated constructor stub
this.regex = regex; } @Override public boolean accept(Path path) { // TODO Auto-generated method stub
boolean flag = path.toString().matches(regex); return !flag; } } /** * * @function 接受 regex 格式的文件 * */
public static class RegexAcceptPathFilter implements PathFilter { private final String regex; public RegexAcceptPathFilter(String regex) { // TODO Auto-generated constructor stub
this.regex = regex; } @Override public boolean accept(Path path) { // TODO Auto-generated method stub
boolean flag = path.toString().matches(regex); return flag; } } /** * @param args * @throws IOException * @throws URISyntaxException */
public static void main(String[] args) throws URISyntaxException, IOException { // TODO Auto-generated method stub
list(); } private static void list() throws URISyntaxException, IOException { // TODO Auto-generated method stub
Configuration conf = new Configuration();//讀取Hadoop配置文件 //設置文件系統訪問接口，並創建FileSystem在本地的運行模式
URI uri = new URI("hdfs://Centpy:9000"); hdfs = FileSystem.get(uri, conf); local = FileSystem.getLocal(conf);//獲取本地文件系統 //過濾目錄下的svn文件
FileStatus[] dirstatus = local.globStatus(new Path("D://Code/EclipseCode/mergeSmallFilesTestData/*"), new RegexExcludePathFilter("^.*svn$")); //獲取D:\Code\EclipseCode\mergeSmallFilesTestData目錄下的所有文件路徑
Path[] dirs = FileUtil.stat2Paths(dirstatus); FSDataOutputStream out = null; FSDataInputStream in = null; for(Path dir:dirs) {//比如拿2018-03-23為例 //將文件夾名稱2018-03-23的-去掉，直接，得到20180323文件夾名稱
String fileName = dir.getName().replace("-", "");//文件名稱 //只接受2018-03-23日期目錄下的.txt文件
FileStatus[] localStatus = local.globStatus(new Path(dir + "/*"), new RegexAcceptPathFilter("^.*txt$")); // 獲得2018-03-23日期目錄下的所有文件
Path[] listPath = FileUtil.stat2Paths(localStatus); // 輸出路徑
Path outBlock = new Path("hdfs://Centpy:9000/mergeSmallFiles/result/"+ fileName + ".txt"); System.out.println("合並后的文件名稱："+fileName+".txt"); // 打開輸出流
out = hdfs.create(outBlock); //循環操作2018-03-23日期目錄下的所有文件
for(Path p:listPath) { in = local.open(p);// 打開輸入流
IOUtils.copyBytes(in, out, 4096, false);// 復制數據
in.close();// 關閉輸入流
} if (out != null) { out.close();// 關閉輸出流
} } } }
測試結果
運行程序之后會將本地D://Code/EclipseCode/mergeSmallFilesTestData路徑下的每個文件夾下的n個.txt文件內容合並到一個.txt文件中，並存放到指定的HDFS路徑（"hdfs://Centpy:9000/mergeSmallFiles/result/"）下。
以上就是博主為大家介紹的這一板塊的主要內容，這都是博主自己的學習過程，希望能給大家帶來一定的指導作用，有用的還望大家點個支持，如果對你沒用也望包涵，有錯誤煩請指出。如有期待可關注博主以第一時間獲取更新哦，謝謝！
版權聲明：本文為博主原創文章，未經博主允許不得轉載。