Java，操作HDFS文件系统，文件的上传、下载和删除完成，代码案例

分布式文件系统

HDFS，分布式文件系统

HDFS，File，分布式文件系统，具有高容错（fault-）特性，被设计为部署在低成本（low-cost）硬件上，并且它提供高吞吐量（）来访问应用程序数据，适合适用于具有非常大数据集（large data set）的应用。 HDFS放宽（relax）了POSIX（）的要求，使得文件系统中的数据可以以流的形式被访问（）。 HDFS 最初是为开源项目 nutch 的基础设施而创建的。 HDFS 是该项目的一部分，也是它的一部分。

HDFS、环境搭建

上一节：构建大数据集群

Java操作HDFS文件系统

pom.xml

    
        org.apache.hadoop
        hadoop-client
        3.1.3
    
    
        junit
        junit
        4.12
    
    
        org.slf4j
        slf4j-log4j12
        1.7.30
    


    11
    11

代码案例1：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import java.io.IOException;
import java.net.URI;
import java.util.Arrays;
public class HdfsDemo {
    public static void main(String[] args) throws Exception {
        FileSystem fileSystem = getHdfs();
        // 创建一个文件夹
        fileSystem.mkdirs(new Path("/demo/hello"));
        // 上传文件
        Path localPath1 = new Path("D:\\Temp\\hello.txt");
        Path remotePath1 = new Path("/demo/hello/");
        fileSystem.copyFromLocalFile(false, true, localPath1, remotePath1);
        Path localPath2 = new Path("D:\\Temp\\hello2.txt");
        Path remotePath2 = new Path("/demo/hello/");
        fileSystem.copyFromLocalFile(false, true, localPath2, remotePath2);
        // 下载文件
        Path remotePath3 = new Path("/demo/hello/hello2.txt");
        Path remotePath4 = new Path("/demo/hello/helloworld2.txt");
        fileSystem.rename(remotePath3, remotePath4);
        // 下载文件
        Path remoteDownPath1 = new Path("/demo/hello/hello.txt");
        Path localDownPath1 = new Path("D:/Temp/hello3.txt");
        // fileSystem.copyToLocalFile(false, remoteDownPath1, localDownPath1, true);
        // 删除文件
        fileSystem.delete(remoteDownPath1, false);
        // 获取文件详细信息
        fileDetail(fileSystem);
        //  判断是文件夹还是文件
        file(fileSystem);
        close(fileSystem);
    }
    public static void file(FileSystem fileSystem) throws IOException {
        FileStatus[] listStatus = fileSystem.listStatus(new Path("/"));
        for (FileStatus status : listStatus) {
            if (status.isFile()) {
                System.out.println("文件:" + status.getPath().getName());
            } else {
                System.out.println("目录:" + status.getPath().getName());
            }
        }
    }
    public static void fileDetail(FileSystem fileSystem) throws IOException {
        // 获取所有文件信息
        RemoteIterator listFiles = fileSystem.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));
        }
    }
    
    public static FileSystem getHdfs() throws Exception {
        // 获取连接集群的地址
        URI uri = new URI("hdfs://192.168.2.53:8020");
        Configuration configuration = new Configuration();
        //设置配置文件中副本的数量
        configuration.set("dfs.replication", "2");
        configuration.set("fs.defaultFS", "hdfs://192.168.2.53:8020");
        // 用户
        String user = "admin";
        FileSystem fileSystem = FileSystem.get(uri, configuration, user);
        return fileSystem;
    }
    
    public static void close(FileSystem fileSystem) throws IOException {
        if (fileSystem != null) {
            fileSystem.close();
        }
        fileSystem = null;
    }
}

代码案例2：

import java.io.FileInputStream;
import java.io.FileOutputStream;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
//操作HDFS
public class FileHdfsDownLoad {
    //下载
    public static void main(String[] args) throws Exception {
        //创建一个操作HDFS的对象
        Configuration config = new Configuration();
        //设置java代码连接哪个hdfs
        config.set("fs.defaultFS", "hdfs://192.168.2.53:8020");
        //获取操作HDFS的对象
        FileSystem fileSystem = FileSystem.get(config);
        //下载文件
        downloadFile(fileSystem);
        // 上传文件
        putFile(fileSystem);
        // 关闭
        fileSystem.close();
    }
    private static void putFile(FileSystem fileSystem) throws Exception {
        //进行上传文件操作
        //获取输入流
        FileInputStream fileInputStream = new FileInputStream("D:/Temp/Anaconda.txt");
        //获取hdfs中的输出流
        Path remoteDownPath1 = new Path("/demo/hello/Anaconda.txt");
        FSDataOutputStream fsDataOutputStream = fileSystem.create(remoteDownPath1);
        //第一个参数是一个输入流、第二个参数是输出流，第三个是缓存区大小，第四个参数是是否关闭
        IOUtils.copyBytes(fileInputStream, fsDataOutputStream, 1024, true);
    }
    private static void downloadFile(FileSystem fileSystem) throws Exception {
        //获取hdfs分布式文件系统中的输入流
        Path remoteDownPath1 = new Path("/demo/hello/hello2.txt");
        FSDataInputStream fsDataInputStream = fileSystem.open(remoteDownPath1);
        //获取本地文件输出流
        FileOutputStream foFileOutputStream = new FileOutputStream("D:/Temp/hello3.txt");
        //下载文件
        IOUtils.copyBytes(fsDataInputStream, foFileOutputStream, 1024, true);
    }
}

查看效果：