013_HDFS文件合并上传putmarge功能（类似于hadoop fs -getmerge）

场景
合并小文件，存放到HDFS上。例如，当需要分析来自许多服务器的Apache日志时，各个日志文件可能比较小，然而Hadoop更合适处理大文件，效率会更高，此时就需要合并分散的文件。如果先将所有文件合并，在复制上传到HDFS上的话，需要占用本地计算机的大量磁盘空间。采取在向HDFS复制上传文件的过程中将小文件进行合并，效果会更好。

开发一个PutMerge程序，用于将合并文件后放入HDFS。

命令getmerge用于将一组HDFS文件在复制到本地计算机一起进行合并。

分析

文件的上传和下载就是字节字符流的读写操作
读文件：输入流-->read
写文件：输出流-->write
1、每个本地文件代开输入流，进行读取内容
2、HDFS文件打开输出流，进行内容写入。
3、循环操作（就是进行把小文件用流读出，在流状态下汇总到一定规模，然后写入到hdfs）
4、关闭流



实现代码

package org.dragon.hadoop.hdfs;

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;

/**
* 功能：在向HDFS上传文件的过程中，进行合并文件
* @author ZhuXY
* @time   2016-3-7 上午9:24:13
*/
public class PutMerge {

/**
* 复制上传文件，并将文件合并
* @param localDir：本地上传的文件目录
* @param hdfsFile：HDFS上传文件的名称，包括路径
*/
public static void putMerge(String localDir,String hdfsFile) {
//1)获取配置信息
Configuration conf=new Configuration();

//2)路径
Path localPath=new Path(localDir);
Path hdfsPath=new Path(hdfsFile);

try {
//获取本地文件系统
FileSystem localFs=FileSystem.getLocal(conf);

//获取HDFS文件系统
FileSystem hdfs=FileSystem.get(conf);

FileStatus[] fileStatus=localFs.listStatus(localPath);

//打开HDFS文件的输出流
FSDataOutputStream outputStream=hdfs.create(hdfsPath);

//循环遍历本地文件
for (FileStatus fiStatus:fileStatus) {
//获取文件
Path path=fiStatus.getPath();

System.out.println("文件为"+path.getName());

//打开文件的输入流
FSDataInputStream inputStream=localFs.open(path);

//进行流的读写操作
byte[] buffer =new byte[1024];

int len=0;
while ((len=inputStream.read(buffer))>0) {
outputStream.write(buffer,0,len);

}
inputStream.close();
}
outputStream.close();
} catch (Exception e) {
e.printStackTrace();
}

}

public static void main(String[] args) {
String localDir="D:/logs";
String hdfsFile="hdfs://hadoop-master.dragon.org:9000/wc/test/logs.data";

putMerge(localDir, hdfsFile);
}
}

有可能出现的问题



其实前面几个博客已经提到，这个需要在jdk1.7环境下编译和运行，jre1.7才行，具体原因不太知道。切换后就好了。