Hadoop 系列（七）—— HDFS Java API

一、 簡介

想要使用 HDFS API，需要導入依賴 hadoop-client。如果是 CDH 版本的 Hadoop，還需要額外指明其倉庫地址：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.heibaiying</groupId>
    <artifactId>hdfs-java-api</artifactId>
    <version>1.0</version>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <hadoop.version>2.6.0-cdh6.15.2</hadoop.version>
    </properties>

    <!---配置 CDH 倉庫地址-->
    <repositories>
        <repository>
            <id>cloudera</id>
            <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
        </repository>
    </repositories>

    <dependencies>
        <!--Hadoop-client-->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

</project>

二、API的使用

2.1 FileSystem

FileSystem 是所有 HDFS 操作的主入口。由于之后的每個單元測試都需要用到它，這里使用 @Before 注解進行標注。

private static final String HDFS_PATH = "hdfs://192.168.0.106:8020";
private static final String HDFS_USER = "root";
private static FileSystem fileSystem;

@Before
public void prepare() {
    try {
        Configuration configuration = new Configuration();
        // 這里我啟動的是單節點的 Hadoop,所以副本系數設置為 1,默認值為 3
        configuration.set("dfs.replication", "1");
        fileSystem = FileSystem.get(new URI(HDFS_PATH), configuration, HDFS_USER);
    } catch (IOException e) {
        e.printStackTrace();
    } catch (InterruptedException e) {
        e.printStackTrace();
    } catch (URISyntaxException e) {
        e.printStackTrace();
    }
}

@After
public void destroy() {
    fileSystem = null;
}

2.2 創建目錄

支持遞歸創建目錄：

@Test
public void mkDir() throws Exception {
    fileSystem.mkdirs(new Path("/hdfs-api/test0/"));
}

2.3 創建指定權限的目錄

FsPermission(FsAction u, FsAction g, FsAction o) 的三個參數分別對應：創建者權限，同組其他用戶權限，其他用戶權限，權限值定義在 FsAction 枚舉類中。

@Test
public void mkDirWithPermission() throws Exception {
    fileSystem.mkdirs(new Path("/hdfs-api/test1/"),
            new FsPermission(FsAction.READ_WRITE, FsAction.READ, FsAction.READ));
}

2.4 創建文件，并寫入內容

@Test
public void create() throws Exception {
    // 如果文件存在，默認會覆蓋, 可以通過第二個參數進行控制。第三個參數可以控制使用緩沖區的大小
    FSDataOutputStream out = fileSystem.create(new Path("/hdfs-api/test/a.txt"),
                                               true, 4096);
    out.write("hello hadoop!".getBytes());
    out.write("hello spark!".getBytes());
    out.write("hello flink!".getBytes());
    // 強制將緩沖區中內容刷出
    out.flush();
    out.close();
}

2.5 判斷文件是否存在

@Test
public void exist() throws Exception {
    boolean exists = fileSystem.exists(new Path("/hdfs-api/test/a.txt"));
    System.out.println(exists);
}

2.6 查看文件內容

查看小文本文件的內容，直接轉換成字符串后輸出：

@Test
public void readToString() throws Exception {
    FSDataInputStream inputStream = fileSystem.open(new Path("/hdfs-api/test/a.txt"));
    String context = inputStreamToString(inputStream, "utf-8");
    System.out.println(context);
}

inputStreamToString 是一個自定義方法，代碼如下：

/**
 * 把輸入流轉換為指定編碼的字符
 *
 * @param inputStream 輸入流
 * @param encode      指定編碼類型
 */
private static String inputStreamToString(InputStream inputStream, String encode) {
    try {
        if (encode == null || ("".equals(encode))) {
            encode = "utf-8";
        }
        BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, encode));
        StringBuilder builder = new StringBuilder();
        String str = "";
        while ((str = reader.readLine()) != null) {
            builder.append(str).append("\n");
        }
        return builder.toString();
    } catch (IOException e) {
        e.printStackTrace();
    }
    return null;
}

2.7 文件重命名

@Test
public void rename() throws Exception {
    Path oldPath = new Path("/hdfs-api/test/a.txt");
    Path newPath = new Path("/hdfs-api/test/b.txt");
    boolean result = fileSystem.rename(oldPath, newPath);
    System.out.println(result);
}

2.8 刪除目錄或文件

public void delete() throws Exception {
    /*
     *  第二個參數代表是否遞歸刪除
     *    +  如果 path 是一個目錄且遞歸刪除為 true, 則刪除該目錄及其中所有文件;
     *    +  如果 path 是一個目錄但遞歸刪除為 false,則會則拋出異常。
     */
    boolean result = fileSystem.delete(new Path("/hdfs-api/test/b.txt"), true);
    System.out.println(result);
}

2.9 上傳文件到HDFS

@Test
public void copyFromLocalFile() throws Exception {
    // 如果指定的是目錄，則會把目錄及其中的文件都復制到指定目錄下
    Path src = new Path("D:\\BigData-Notes\\notes\\installation");
    Path dst = new Path("/hdfs-api/test/");
    fileSystem.copyFromLocalFile(src, dst);
}

2.10 上傳大文件并顯示上傳進度

@Test
    public void copyFromLocalBigFile() throws Exception {

        File file = new File("D:\\kafka.tgz");
        final float fileSize = file.length();
        InputStream in = new BufferedInputStream(new FileInputStream(file));

        FSDataOutputStream out = fileSystem.create(new Path("/hdfs-api/test/kafka5.tgz"),
                new Progressable() {
                  long fileCount = 0;

                  public void progress() {
                     fileCount++;
                     // progress 方法每上傳大約 64KB 的數據后就會被調用一次
                     System.out.println("上傳進度：" + (fileCount * 64 * 1024 / fileSize) * 100 + " %");
                   }
                });

        IOUtils.copyBytes(in, out, 4096);

    }

2.11 從HDFS上下載文件

@Test
public void copyToLocalFile() throws Exception {
    Path src = new Path("/hdfs-api/test/kafka.tgz");
    Path dst = new Path("D:\\app\\");
    /*
     * 第一個參數控制下載完成后是否刪除源文件,默認是 true,即刪除;
     * 最后一個參數表示是否將 RawLocalFileSystem 用作本地文件系統;
     * RawLocalFileSystem 默認為 false,通常情況下可以不設置,
     * 但如果你在執行時候拋出 NullPointerException 異常,則代表你的文件系統與程序可能存在不兼容的情況 (window 下常見),
     * 此時可以將 RawLocalFileSystem 設置為 true
     */
    fileSystem.copyToLocalFile(false, src, dst, true);
}

2.12 查看指定目錄下所有文件的信息

public void listFiles() throws Exception {
    FileStatus[] statuses = fileSystem.listStatus(new Path("/hdfs-api"));
    for (FileStatus fileStatus : statuses) {
        //fileStatus 的 toString 方法被重寫過，直接打印可以看到所有信息
        System.out.println(fileStatus.toString());
    }
}

FileStatus 中包含了文件的基本信息，比如文件路徑，是否是文件夾，修改時間，訪問時間，所有者，所屬組，文件權限，是否是符號鏈接等，輸出內容示例如下：

FileStatus{
path=hdfs://192.168.0.106:8020/hdfs-api/test; 
isDirectory=true; 
modification_time=1556680796191; 
access_time=0; 
owner=root; 
group=supergroup; 
permission=rwxr-xr-x; 
isSymlink=false
}

2.13 遞歸查看指定目錄下所有文件的信息

@Test
public void listFilesRecursive() throws Exception {
    RemoteIterator<LocatedFileStatus> files = fileSystem.listFiles(new Path("/hbase"), true);
    while (files.hasNext()) {
        System.out.println(files.next());
    }
}

和上面輸出類似，只是多了文本大小，副本系數，塊大小信息。

LocatedFileStatus{
path=hdfs://192.168.0.106:8020/hbase/hbase.version; 
isDirectory=false; 
length=7; 
replication=1; 
blocksize=134217728; 
modification_time=1554129052916; 
access_time=1554902661455; 
owner=root; group=supergroup;
permission=rw-r--r--; 
isSymlink=false}

2.14 查看文件的塊信息

@Test
public void getFileBlockLocations() throws Exception {

    FileStatus fileStatus = fileSystem.getFileStatus(new Path("/hdfs-api/test/kafka.tgz"));
    BlockLocation[] blocks = fileSystem.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
    for (BlockLocation block : blocks) {
        System.out.println(block);
    }
}

塊輸出信息有三個值，分別是文件的起始偏移量 (offset)，文件大小 (length)，塊所在的主機名 (hosts)。

0,57028557,hadoop001

這里我上傳的文件只有 57M(小于 128M)，且程序中設置了副本系數為 1，所有只有一個塊信息。

<br/>

<br/>

以上所有測試用例下載地址：HDFS Java API

更多大數據系列文章可以參見 GitHub 開源項目： 大數據入門指南