Hadoop文件合并——Hadoop In Action上的一個示例

    上一篇文章已經詳細的說明了如何在Eclipse下面遠程連接Hadoop集群，進行Hadoop程序開發。這里說明一個Hadoop In Action書上的一個示例，可能是由于Hadoop版本更新的問題，導致樹上的一些個示例程序沒有辦法正常執行。

    整個代碼的工作就是把本地目錄下個若干個小文件，合并成一個較大的文件，寫入到HDFS中。話不多說，代碼如下：

    補充說明：后來發現，書上的源代碼是沒有問題的，只不過是書上的源代碼要打成jar包，放在集群的機器上去運行，如果在Eclipse下面調試運行的話，就會出現問題。出現問題的原因如下

//讀取本地文件系統，如果要想正確運行，必須要打成jar包，在hadoop集群的機器上面運行
FileSystem hdfs = FileSystem.get(conf);
FileSystem local = FileSystem.getLocal(conf);

//通過URI可以遠程讀取HDFS，所以Eclipse下面調試要使用這種寫法，打成jar包這種形式也是可以的
FileSystem hdfs = FileSystem.get(URI.create(serverPath), conf);
FileSystem local = FileSystem.getLocal(conf);

package com.hadoop.examples;
import java.io.IOException;
import java.net.URI;

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;

/**
 * @Package
 * @ClassName: PutMerge
 * @Description: 讀取本地目錄下的文件，寫入到HDFS，在寫入的過程中，
 *                 把這三個文件合成一個文件
 * @author lxy
 * @date 2015年3月25日 上午9:59:38
 * @version V1.0
 */

public class PutMerge {

    public static void main(String[] args) throws IOException {

        // 輸入目錄，目錄下有三個txt，文章最后面會兒給出文件內容
        String localPathStr = "E:\\test";
        // 輸出目錄，HDFS路徑，文章最后面會給出合并之后的文件內容
        String serverPath =
                "hdfs://192.168.3.57:8020/user/lxy/mergeresult/merge.txt";
        
        //輸入目錄，是一個本地目錄
        Path inputDir = new Path(localPathStr);
        //輸出目錄，是一個HDFS路徑
        Path hdfsFile = new Path(serverPath);

        Configuration conf = new Configuration();

        /**
         * Hadoop in Action的原代碼如下 
         * FileSystem hdfs = FileSystem.get(conf);
         * 但是這樣的話，使用Eclipse調試時，執行下面的語句是就會報異常，因為它是讀取本地
         * 文件系統
         *  FSDataOutputStream out = hdfs.create(hdfsFile);
         */
        // 根據上面的serverPath，獲取到的是一個org.apache.hadoop.hdfs.DistributedFileSystem對象
        FileSystem hdfs = FileSystem.get(URI.create(serverPath), conf);
        FileSystem local = FileSystem.getLocal(conf);

        try {
            //獲取輸入目錄下的文件以及文件夾列表
            FileStatus[] inputFiles = local.listStatus(inputDir);
            //在hdfs上創建一個文件
            FSDataOutputStream out = hdfs.create(hdfsFile);

            for (int i = 0; i < inputFiles.length; i++) {
                System.out.println(inputFiles[i].getPath().getName());
                //打開本地輸入流
                FSDataInputStream in = local.open(inputFiles[i].getPath());
                byte buffer[] = new byte[256];
                int bytesRead = 0;
                while ((bytesRead = in.read(buffer)) > 0) {
                    //往hdfs上的文件寫數據
                    out.write(buffer, 0, bytesRead);
                }
                //釋放資源
                in.close();
            }
            //釋放資源
            out.close();
        }
        catch (IOException e) {
            e.printStackTrace();
        }
    }
}

    我的測試目錄下有三個txt文件

    1.txt

1 hello Hadoop
2 hello Hadoop
3 hello Hadoop
4 hello Hadoop
5 hello Hadoop
6 hello Hadoop
7 hello Hadoop

    2.txt

8 hello Hadoop
9 hello Hadoop
10 hello Hadoop
11 hello Hadoop
12 hello Hadoop
13 hello Hadoop
14 hello Hadoop

    3.txt

15 hello Hadoop
16 hello Hadoop
17 hello Hadoop
18 hello Hadoop
19 hello Hadoop
20 hello Hadoop
21 hello Hadoop

    合并之后的文件如下所示：