项目背景
在实际项目中，输入数据往往是由许多小文件组成，这里的小文件是指小于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/"）下。
以上就是博主为大家介绍的这一板块的主要内容，这都是博主自己的学习过程，希望能给大家带来一定的指导作用，有用的还望大家点个支持，如果对你没用也望包涵，有错误烦请指出。如有期待可关注博主以第一时间获取更新哦，谢谢！
版权声明：本文为博主原创文章，未经博主允许不得转载。