HDFS

HDFS，分布式文件系统

HDFS，Hadoop Distributed File System，分布式文件系统，有着高容错性（fault-tolerent）的特点，并且设计用来部署在低廉的（low-cost）硬件上，而且它提供高吞吐量（high throughput）来访问应用程序的数据，适合那些有着超大数据集（large data set）的应用程序。HDFS 放宽了（relax）POSIX 的要求（requirements）这样可以实现流的形式访问（streaming access）文件系统中的数据，HDFS 开始是为开源的 apache 项目nutch的基础结构而创建，HDFS是 hadoop项目的一部分，而hadoop又是lucene的一部分。

HDFS，环境搭建

上一节：大数据集群Hadoop搭建

Java操作HDFS文件系统

pom.xml

<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>
<properties>
    <maven.compiler.source>11</maven.compiler.source>
    <maven.compiler.target>11</maven.compiler.target>
</properties>

代码案例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<LocatedFileStatus> 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));
        }
    }

    /**
     * @return
     * @throws Exception
     */
    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;
    }

    /**
     * @param fileSystem
     * @throws IOException
     */
    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);
    }
}

查看效果：