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才行，具体原因不太知道。切换后就好了。