Hadoop MapReduce编程 API入门系列之小文件合并（二十九）

　　不多说，直接上代码。

Hadoop 自身提供了几种机制来解决相关的问题，包括HAR，SequeueFile和CombineFileInputFormat。

Hadoop 自身提供的几种小文件合并机制

Hadoop HAR

将众多小文件打包成一个大文件进行存储，并且打包后原来的文件仍然可以通过Map-reduce进行操作，打包后的文件由索引和存储两大部分组成

        缺点：一旦创建就不能修改，也不支持追加操作，还不支持文档压缩，当有新文件进来以后，需要重新打包。

SequeuesFile

Sequence file由一系列的二进制key/value组成，如果key为小文件名，value为文件内容，则可以将大批小文件合并成一个大文件。

        优缺点：对小文件的存取都比较*，也不限制用户和文件的多少，但是该方法不能使用append方法，所以适合一次性写入大量小文件的场景。

CombineFileInputFormat

CombineFileInputFormat是一种新的inputformat，用于将多个文件合并成一个单独的split作为输入，而不是通常使用一个文件作为输入。另外，它会考虑数据的存储位置。

目前很多公司采用的方法就是在数据进入 Hadoop 的 HDFS 系统之前进行合并（也是本博文这方法），一般效果较上述三种方法明显。

 代码

package zhouls.bigdata.myMapReduce.MergeSmallFiles;

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;
/**
* function 合并小文件至 HDFS
*
*
*/
public class MergeSmallFilesToHDFS {
private static FileSystem fs = null;
private static FileSystem local = null;
/**
* @function main
* @param args
* @throws IOException
* @throws URISyntaxException
*/
public static void main(String[] args) throws IOException,
URISyntaxException {
list();
}

/**
*
* @throws IOException
* @throws URISyntaxException
*/
public static void list() throws IOException, URISyntaxException {
// 读取hadoop文件系统的配置
Configuration conf = new Configuration();
//文件系统访问接口
URI uri = new URI("hdfs://HadoopMaster:9000");
//创建FileSystem对象aa
fs = FileSystem.get(uri, conf);
// 获得本地文件系统
local = FileSystem.getLocal(conf);
//过滤目录下的 svn 文件
FileStatus[] dirstatus = local.globStatus(new Path("./data/mergeSmallFiles/*"),new RegexExcludePathFilter("^.*svn$"));
//获取73目录下的所有文件路径
Path[] dirs = FileUtil.stat2Paths(dirstatus);
FSDataOutputStream out = null;
FSDataInputStream in = null;
for (Path dir : dirs) {
String fileName = dir.getName().replace("-", "");//文件名称
//只接受日期目录下的.txt文件a
FileStatus[] localStatus = local.globStatus(new Path(dir+"/*"),new RegexAcceptPathFilter("^.*txt$"));
// 获得日期目录下的所有文件
Path[] listedPaths = FileUtil.stat2Paths(localStatus);
//输出路径
Path block = new Path("hdfs://HadoopMaster:9000/tv/"+ fileName + ".txt");
// 打开输出流
out = fs.create(block);
for (Path p : listedPaths) {
in = local.open(p);// 打开输入流
IOUtils.copyBytes(in, out, 4096, false); // 复制数据
// 关闭输入流
in.close();
}
if (out != null) {
// 关闭输出流a
out.close();
}
}

}

/**
*
* @function 过滤 regex 格式的文件
*
*/
public static class RegexExcludePathFilter implements PathFilter {
private final String regex;

public RegexExcludePathFilter(String regex) {
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) {
this.regex = regex;
}

@Override
public boolean accept(Path path) {
// TODO Auto-generated method stub
boolean flag = path.toString().matches(regex);
return flag;
}

}
}