HDFS 的 API 操作

一、环境准备

1. 配置 windows 的环境变量 

　　在此处下载 windows 的依赖项 https://gitee.com/parasol_ry/hadoop/tree/master/hadoop-3.1.0/bin

　　配置环境变量：

　　

　　 

 　在 IDEA 中创建一个 Maven 工程 HdfsClientDemo，并导入相应的依赖坐标+日志添加

<dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>3.1.3</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.30</version>
        </dependency>
</dependencies>

　　在项目的 src/main/resources 目录下，新建一个文件，命名为“log4j.properties”，在文件 中填入

log4j.rootLogger=INFO, stdout 
log4j.appender.stdout=org.apache.log4j.ConsoleAppender 
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout 
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n 
log4j.appender.logfile=org.apache.log4j.FileAppender 
log4j.appender.logfile.File=target/spring.log 
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout 
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

　　HDFS 的 API 案例实操 

package com.learn.hdfs;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Arrays;

public class HdfsClient {

    //升级为全局变量是 ctrl + alt + F
    // ctrl + p 查看参数有哪些
    private FileSystem fs;

    @Before
    public void init() throws URISyntaxException, IOException, InterruptedException {
        //连接的nn地址
        URI uri = new URI("hdfs://node01:8020");

        //配置一个文件
        Configuration configuration = new Configuration();
        configuration.set("dfs.replication", "2");

        //用户
        String user = "root";

        //获取客户端对象
        fs = FileSystem.get(uri, configuration, user);


    }

    @After
    public void close() throws IOException {
        //关闭资源
        fs.close();
    }

    //创建目录
    @Test
    public void testmkdir() throws URISyntaxException, IOException, InterruptedException {
        //创建一个文件夹
        fs.mkdirs(new Path("/xiyou/huaguoshan3"));
    }

    /*
     *参数优先级
     * hdfs-default.xml => hdfs-site.xml => 在项目资源目录下的配置文件优先级 => 代码中的优先级配置
     * 优先级由 低 到 高
     * */

    //上传
    @Test
    public void testPut() throws IOException {
        //参数解读：
        // 1. 删除原始数据 2.是否允许覆盖 3.原始数据路径 4.目的地路径
        fs.copyFromLocalFile(false, true, new Path("D:\hadoop\sun.txt"), new Path("/xiyou/huaguoshan"));
    }


    //文件下载
    @Test
    public void testGet() throws IOException {
        //参数解读
        // 1. 源文件是否删除 2.原文件的路径 3.目标地址路径 4.是否开启校验
        fs.copyToLocalFile(false, new Path("/xiyou/huaguoshan"), new Path("D:\hadoop"), true);
    }

    //删除
    @Test
    public void testRm() throws IOException {
        //参数解读
        // 1. 要删除的路径 2. 是否递归删除
        // 删除文件
        fs.delete(new Path("/jdk-8u291-linux-x64.tar.gz"), false);

        //删除目录
        fs.delete(new Path("/xiyou/huaguoshan"), true);
    }

    //文件的更名和移动
    @Test
    public void testMv() throws IOException {
        //参数解读
        // 1.原文件的路径 2.目标文件的路径
        //对文件的修改
        fs.rename(new Path("/input/word.txt"), new Path("/input/ss.txt"));
    }

    //获取文件详细信息
    @Test
    public void fileDetail() throws IOException {
        //参数解读
        // 1.文件的路径 2.是否递归遍历
        //获取所有文件信息                 迭代器
        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);

        //遍历文件
        while (listFiles.hasNext()) {
            LocatedFileStatus fileStatus = listFiles.next();

            System.out.println("=======" + fileStatus.getPath() + "=======");
            System.out.println(fileStatus.getPermission());
            System.out.println(fileStatus.getOwner());
            System.out.println(fileStatus.getGroup());
            System.out.println(fileStatus.getLen());
            System.out.println(fileStatus.getModificationTime());
            System.out.println(fileStatus.getReplication());
            System.out.println(fileStatus.getBlockSize());
            System.out.println(fileStatus.getPath().getName());


            //获取块信息
            BlockLocation[] blockLocations = fileStatus.getBlockLocations();
            System.out.println(Arrays.toString(blockLocations));
        }
    }

    //判断是文件还是文件夹
    @Test
    public void testFile() throws IOException {
        FileStatus[] listStatus = fs.listStatus(new Path("/"));

        for (FileStatus status : listStatus) {
            if (status.isFile()) {
                System.out.println("文件: " + status.getPath().getName());
            } else {
                System.out.println("路径: " + status.getPath().getName());
            }
        }
    }
}