Java实现对HDFS文件系统的基本操作

1.准备好jar包

2.创建一个类

1. 测试连接

    @Test   //测试是否连接成功
public void test() {
    //添加配置 ==> core-site.xml
    Configuration conf = new Configuration();
    //配置默认地址端口
    conf.set("fs.defaultFS", "hdfs://192.168.1.105:9000");
    try {
        //加载配置
        FileSystem fs = FileSystem.get(conf);
        //获取目标状态   注意：要用绝对路径，且路径都要用Path包起来
        FileStatus fst = fs.getFileStatus(new Path("/hello"));
        System.out.println(fst.isDirectory());  //是否是文件夹
        System.out.println(fst.isFile());  // 是否是文件
        System.out.println(fst.getLen());  // 获取长度
        System.out.println(fst.getPath()); // 获取路径
                    fs.close();
    } catch (IOException e) {
        e.printStackTrace();
    }
}

2. 在文件系统下创建文件夹

    @Test   //hdfs文件系统中创建文件夹
public void mkdirsFile() {
    //添加配置 ==> core-site.xml
    Configuration conf = new Configuration();
    //配置默认地址端口
    conf.set("fs.defaultFS", "hdfs://192.168.1.105:9000");
    try {
        //加载配置
        FileSystem fs = FileSystem.get(conf);
        //创建文件夹   注意：要用绝对路径，且路径都要用Path包起来
        boolean mkdirs = fs.mkdirs(new Path("/other/test/jdk"));
        System.out.print(mkdirs);
                    fs.close();
    } catch (IOException e) {
        e.printStackTrace();
    }
}

测试时，发现出错了，用户Administator在hadoop上执行写操作时被权限系统拒绝，百度（https://blog.csdn.net/xiaoshunzi111/article/details/52062640）发现，解决有三个办法：
1、在hdfs的配置文件中，将dfs.permissions修改为False
2、执行这样的操作 hadoop fs -chmod 777 /user/hadoop
3、在系统的环境变量里面添加HADOOP_USER_NAME=root（HDFS上的有权限的用户，具体看自己的情况），插入代码实现 System.setProperty("HADOOP_USER_NAME", "root");
前两个不太安全，我选用第三个，重启就可以。

3.递归查看文件夹

    @Test   //hdfs文件系统中遍历文件夹
public void ls() {
    Configuration conf = new Configuration();
    conf.set("fs.defaultFS", "hdfs://192.168.1.105:9000");
    try {
        FileSystem fs = FileSystem.get(conf);
        FileStatus[] fls = fs.listStatus(new Path("/"));
        for(FileStatus fst:fls) {
            judge(fs,fst);
        }
        fs.close();
    } catch (IOException e) {
        e.printStackTrace();
    }
}

public void judge(FileSystem fs,FileStatus fst) {
    String name = fst.getPath().toString().split("hdfs://192.168.1.105:9000/")[1];
    if(fst.isDirectory()) {
        System.out.println("d: "+name);
        try {
            FileStatus[] fls = fs.listStatus(new Path("/"+name));
            for(FileStatus fst2:fls) {
                judge(fs,fst2);
            }
        } catch (IllegalArgumentException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }else {
        System.out.println("f: "+name);
    }
}

4.上传

    @Test   //上传文件到hdfs文件系统    通过io流
public void rz() {
            System.setProperty("HADOOP_USER_NAME", "root");
    Configuration conf = new Configuration();
    conf.set("fs.defaultFS", "hdfs://192.168.1.105:9000");
    try {
        FileSystem fs = FileSystem.get(conf);
        //输出流
        FSDataOutputStream create = fs.create(new Path("/other/hadoop-2.7.7.tar.gz"));
        //输入流
        FileInputStream fis = new FileInputStream(new File("F:\\Linux\\hadoop-2.7.7.tar.gz"));
        int len = 0;
        //定义一个byte类型的数组,数组的大小表示每次从文件中读取出来的数据量
        byte[] b = new  byte[1024];
        //循环读取数据，如果fis.read没有读到就数据返回-1
        while((len = fis.read(b))!=-1) {
            //输入到输出
            create.write(b, 0, len);
        }
        fis.close();
        create.close();
        fs.close();
    } catch (IOException e) {
        e.printStackTrace();
    }
}

@Test   //上传文件到hdfs文件系统    通过核心类FileSystem提供的方法
public void rz1() {
            System.setProperty("HADOOP_USER_NAME", "root");
    Configuration conf = new Configuration();
    conf.set("fs.defaultFS", "hdfs://192.168.1.105:9000");
    try {
        FileSystem fs = FileSystem.get(conf);
        fs.copyFromLocalFile(new Path("F:\\Linux\\hadoop-2.7.7.tar.gz"), new Path("/other/hadoop-2.7.7.tar.gz"));
        fs.close();
    } catch (IOException e) {
        e.printStackTrace();
    }
}

5.下载

    @Test   //下载到本地    通过io流
public void sz() {
    Configuration conf = new Configuration();
    conf.set("fs.defaultFS", "hdfs://192.168.1.105:9000");
    try {
        FileSystem fs = FileSystem.get(conf);
        FSDataInputStream create = fs.open(new Path("/other/hadoop-2.7.7.tar.gz"));
        FileOutputStream fis = new FileOutputStream(new File("F:\\hadoop-2.7.7.tar.gz"));
        int len = 0;
        byte[] b = new  byte[1024];
        while((len = create.read(b))!=-1) {
            fis.write(b, 0, len);
        }
        fis.close();
        create.close();
        fs.close();
    } catch (IOException e) {
        e.printStackTrace();
    }
}

@Test   //下载到本地    通过核心类FileSystem提供的方法
public void sz1() {
    Configuration conf = new Configuration();
    conf.set("fs.defaultFS", "hdfs://192.168.1.105:9000");
    try {
        FileSystem fs = FileSystem.get(conf);
        fs.copyToLocalFile(new Path("/other/hadoop-2.7.7.tar.gz"),new Path("F:\\hadoop-2.7.7.tar.gz"));
        fs.close();
    } catch (IOException e) {
        e.printStackTrace();
    }
}