使用Hadoop提供的API操作HDFS

    本篇文章主要是記錄一下如何使用Hadoop提供的API，通過編程的方式來對HDFS進行增刪查改等操作。Hadoop的版本不同，可能函數的參數會有所不同，本文所使用的是Hadoop 2.5.2。

    整個代碼包含下面幾個方法

1. 1. 讀取HDFS某個文件夾的所有文件，并打印其內容

   2. 重命名HDFS上的文件夾或者文件

   3. 從HDFS下載文件或者文件夾的所有內容到本地

   4. 從本地上傳文件或者文件夾到HDFS

   5. 在HDFS上創建文件夾

   6. 在HDFS上刪除文件夾

   7. 在HDFS上創建文件

   8. 在HDFS上刪除文件

    工程代碼如下，每個方法上面都有相應的注釋，每一個方法都是經過測試可運行的

package com.hadoop.hdfs;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

/**
 * @Package com.hadoop.hdfs
 * @ClassName: OperateHDFS
 * @Description: 通過Java API操作HDFS文件
 * @author lxy
 * @date 2015年3月25日 下午1:43:00
 * @version V1.0
 */

public class OperateHDFS {

    /**
     * 加載配置文件
     */
    static Configuration conf = new Configuration();

    /**
     * 讀取HDFS某個文件夾的所有文件，并打印
     * 
     * @param hdfsPath
     *            讀取HDFS哪個目錄下的文件，HDFS目錄
     * @throws IOException
     */
    public static void readHDFSAll(String hdfsPath) throws IOException {
        InputStream in = null;

        // 讀取HDFS上的文件系統，獲取到的是一個org.apache.hadoop.hdfs.DistributedFileSystem對象
        FileSystem hdfs = FileSystem.get(URI.create(hdfsPath), conf);

        // 使用緩沖流，進行按行讀取的功能
        BufferedReader buff = null;

        // 獲取要讀取的文件的根目錄
        Path listFiles = new Path(hdfsPath);

        // 獲取要讀取的文件的根目錄的所有二級子文件目錄
        FileStatus stats[] = hdfs.listStatus(listFiles);

        for (int i = 0; i < stats.length; i++) {
            // 判斷是不是目錄，如果是目錄，遞歸調用
            if (stats[i].isDirectory()) {
                readHDFSAll(stats[i].getPath().toString());
            }
            else {
                System.out.println("文件路徑名:" + stats[i].getPath().toString());
                // 獲取Path
                Path p = new Path(stats[i].getPath().toString());
                // 打開文件流
                in = hdfs.open(p);
                buff = new BufferedReader(new InputStreamReader(in));
                String str = null;
                while ((str = buff.readLine()) != null) {
                    System.out.println(str);
                }
                buff.close();
                in.close();
            }
        }
    }

    /**
     * 重命名HDFS上的文件夾或者文件
     * 
     * @param srcPath
     *            要修改的HDFS的文件或者文件夾，HDFS目錄
     * @param destPath
     *            修改成什么文件或者文件夾，HDFS目錄
     * @throws IOException
     */
    public static void renameFileOrDirectoryOnHDFS(String srcPath,
            String destPath) throws IOException {
        // 讀取HDFS上的文件系統
        FileSystem hdfs = FileSystem.get(URI.create(srcPath), conf);

        Path src = new Path(srcPath);
        Path dest = new Path(destPath);

        hdfs.rename(src, dest);
        // 釋放資源
        hdfs.close();

        System.out.println("重命名成功");

    }

    /**
     * 從HDFS下載文件或者文件夾的所有內容到本地
     * 
     * @param downloadPath
     *            下載哪個文件，或者哪個目錄下的所有文件，HDFS目錄
     * @param localPath
     *            下載文件到哪里，本地目錄
     * @throws IOException
     */
    public static void downloadFileorDirectoryOnHDFS(String downloadPath,
            String localPath) throws IOException {
        // 讀取HDFS上的文件系統
        FileSystem hdfs = FileSystem.get(URI.create(downloadPath), conf);

        // 要從哪里下載
        Path download = new Path(downloadPath);
        // 下載到本地的哪個目錄
        Path localDir = new Path(localPath);

        // 從HDFS拷貝到本地目錄下
        hdfs.copyToLocalFile(download, localDir);
        // 釋放資源
        hdfs.close();

        System.out.println("下載成功");
    }

    /**
     * 從本地上傳文件或者文件夾到HDFS
     * 
     * @param uploadPath
     *            上傳哪個目錄下的文件或者文件夾，本地目錄
     * @param localPath
     *            上傳文件到哪里，HDFS目錄
     * @throws Exception
     */
    public static void uploadFileorDirectoryToHDFS(String uploadPath,
            String destPath) throws IOException {
        // 讀取HDFS上的文件系統
        FileSystem hdfs = FileSystem.get(URI.create(destPath), conf);

        // 本地文件或者目錄
        Path localDir = new Path(uploadPath);

        // 要上傳到哪里
        Path dest = new Path(destPath);

        // 從本地目錄拷貝到HDFS目錄下，沒有目錄，會自動創建目錄
        hdfs.copyFromLocalFile(localDir, dest);
        // 釋放資源
        hdfs.close();

        System.out.println("上傳成功");
    }

    /**
     * 在HDFS上創建文件夾
     * 
     * @param pathName
     *            文件夾的路徑，HDFS目錄
     * @throws IOException
     */
    public static void createDirectoryOnHDFS(String pathName)
            throws IOException {
        // 讀取HDFS上的文件系統
        FileSystem hdfs = FileSystem.get(URI.create(pathName), conf);
        Path newPath = new Path(pathName);
        // 在HDFS上創建文件夾
        hdfs.mkdirs(newPath);
        // 釋放資源
        hdfs.close();
        System.out.println("創建文件夾成功");
    }

    /**
     * 在HDFS上刪除文件夾
     * 
     * @param pathName
     *            文件夾的路徑，HDFS目錄
     * @throws IOException
     */
    public static void deleteDirectoryOnHDFS(String deletePath)
            throws IOException {
        // 讀取HDFS上的文件系統
        FileSystem hdfs = FileSystem.get(URI.create(deletePath), conf);
        Path deleteDir = new Path(deletePath);
        // 在HDFS上刪除文件夾
        hdfs.deleteOnExit(deleteDir);
        // 釋放資源
        hdfs.close();
        System.out.println("刪除文件夾成功");
    }

    /**
     * 在HDFS上創建文件
     * 
     * @param filePath
     *            創建的文件的名稱，HDFS目錄
     * @throws IOException
     */
    public static void createFileOnHDFS(String filePath) throws IOException {
        FileSystem hdfs = FileSystem.get(URI.create(filePath), conf);
        Path newFile = new Path(filePath);
        hdfs.createNewFile(newFile);
        // 釋放資源
        hdfs.close();
        System.out.println("創建文件成功");
    }

    /**
     * 在HDFS上刪除文件
     * 
     * @param filePath
     *            刪除的文件的名稱，HDFS目錄
     * @throws IOException
     */
    public static void deleteFileOnHDFS(String filePath) throws IOException {
        FileSystem hdfs = FileSystem.get(URI.create(filePath), conf);
        Path deleteFile = new Path(filePath);
        hdfs.deleteOnExit(deleteFile);
        hdfs.close();
        System.out.println("刪除文件成功");
    }

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

        // 讀取一個目錄下面的所有的文件包括子目錄下的文件
        readHDFSAll("hdfs://192.168.3.57:8020/user/lxy/input");

        // 重命名HDFS上的文件
        renameFileOrDirectoryOnHDFS(
                "hdfs://192.168.3.57:8020/user/lxy/mergeresult/merge.txt",
                "hdfs://192.168.3.57:8020/user/lxy/mergeresult/renamemerge.txt");

        // 重命名HDFS上的文件夾
        renameFileOrDirectoryOnHDFS(
                "hdfs://192.168.3.57:8020/user/lxy/mergeresult",
                "hdfs://192.168.3.57:8020/user/lxy/rename");

        // 從HDFS下載文件或者文件夾
        downloadFileorDirectoryOnHDFS(
                "hdfs://192.168.3.57:8020/user/lxy/mergeresult",
                "E:\\test\\download");
        // 從本地上傳文件到HDFS
        uploadFileorDirectoryToHDFS("E:\\test\\upload\\lib",
                "hdfs://192.168.3.57:8020/user/lxy");

        // 在HDFS上創建文件夾createDir
        createDirectoryOnHDFS("hdfs://192.168.3.57:8020/user/lxy/createDir");

        // 在HDFS上刪除文件夾createDir
        deleteDirectoryOnHDFS("hdfs://192.168.3.57:8020/user/lxy/createDir");

        // 在HDFS上創建文件newFile.txt
        createFileOnHDFS("hdfs://192.168.3.57:8020/user/lxy/newFile.txt");

        // 在HDFS上刪除文件newFile.txt
        deleteFileOnHDFS("hdfs://192.168.3.57:8020/user/lxy/newFile.txt");
    }
}