HDFS的API操作

HDFS的API操作

在 Java 中操作 HDFS, 主要涉及以下 Class

1. Configuration：该类的对象封装了客户端或者服务器的配置
2. FileSystem：该类的对象是一个文件系统对象，可以用该对象的一些方法来对文件进行操作, 通过FileSystem的静态方法 get 获得该对象

/*
在 Java 中操作 HDFS, 主要涉及以下 Class
1. Configuration：该类的对象封装了客户端或者服务器的配置
2. FileSystem：该类的对象是一个文件系统对象, 可以用该对象的一些方法来对文件进行操作, 通过FileSystem 的静态方法 get 获得该对象
 */

import org.apache.commons.compress.utils.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;

import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

public class HdfsClientOpt {
    public static void main(String[] args) throws Exception {
        HdfsClientOpt client = new HdfsClientOpt();
        client.getFileSystem();
    }

    //=================================================================//
    // 获取FileSystem

    /**
     * 方式一：set+get
     *
     * @throws IOException
     */
    public void getFileSystem() throws IOException {
        //1.创建一个Configuration实例化对象
        Configuration configuration = new Configuration();
        //2.设置文件系统类型
        configuration.set("fs.defaultFS", "hdfs://hadoop10:8020");
        //3.获取文件系统
        FileSystem fileSystem = FileSystem.get(configuration);
        //4.打印输出
        System.out.println(fileSystem);
    }

    /**
     * 方式二：set方式+通过newInstance
     *
     * @throws IOException
     */
    public void getFileSystem2() throws IOException {
        //1:创建Configuration对象
        Configuration conf = new Configuration();
        //2:设置文件系统类型
        conf.set("fs.defaultFS", "hdfs://hadoop10:8020");
        //3:获取指定文件系统
        FileSystem fileSystem = FileSystem.newInstance(conf);
        //4:输出测试
        System.out.println(fileSystem);
    }

    /**
     * 方式三：new URI+get
     *
     * @throws Exception
     */
    public void getFileSystem3() throws Exception {
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://hadoop10:8020"), new Configuration());
        System.out.println("fileSystem:" + fileSystem);
    }

    /**
     * 方式四：newInstance+get
     *
     * @throws Exception
     */
    public void getFileSystem4() throws Exception {
        FileSystem fileSystem = FileSystem.newInstance(new
                URI("hdfs://hadoop10:8020"), new Configuration());
        System.out.println("fileSystem:" + fileSystem);
    }

    //=================================================================//
    //文件的遍历

    /**
     * hdfs文件的遍历
     *
     * @throws URISyntaxException
     * @throws IOException
     */
    public void listFiles() throws URISyntaxException, IOException {
        //1、获取FileSystem实例
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://hadoop10:8020"),
                new Configuration());
        //2、调用方法listFiles 获取 /目录下所有的文件信息
        RemoteIterator<LocatedFileStatus> iterator = fileSystem.listFiles(new
                Path("/"), true);
        //3、遍历迭代器
        while (iterator.hasNext()) {
            LocatedFileStatus fileStatus = iterator.next();
            //获取文件的绝对路径 : hdfs://hadoop10:8020/xxx
            System.out.println(fileStatus.getPath() + "======"
                    + fileStatus.getPath().getName());
            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());
            //文件的block信息
            BlockLocation[] blockLocations = fileStatus.getBlockLocations();
            for (BlockLocation blockLocation : blockLocations) {
                String[] hosts = blockLocation.getHosts();
                System.out.println("block数量为: " + hosts.length);
                for (String host : hosts) {
                    System.out.println("主机为: " + host);
                }
            }
            System.out.println();
        }
    }


    //=================================================================//
    //创建文件夹

    /**
     * 创建文件夹
     *
     * @throws URISyntaxException
     * @throws IOException
     * @throws InterruptedException
     */
    public void mkdirs() throws URISyntaxException, IOException,
            InterruptedException {
        //1:获取FileSystem实例
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://hadoop10:8020"),
                new Configuration(), "root");
        //2:创建文件夹
        boolean bl = fileSystem.mkdirs(new Path("/aaa/bbb/ccc"));
        System.out.println(bl);
        //3: 关闭FileSystem
        fileSystem.close();
    }


    //=================================================================//
    //文件的上传

    /**
     * 文件的上传
     *
     * @throws URISyntaxException
     * @throws IOException
     * @throws InterruptedException
     */
    public void uploadFile() throws URISyntaxException, IOException,
            InterruptedException {
        //1:获取FileSystem
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://hadoop10:8020"),
                new Configuration(), "root");
        //2:调用方法，实现上传
        fileSystem.copyFromLocalFile(new Path("D://test1.txt"), new Path("/"));
        //3:关闭FileSystem
        fileSystem.close();
    }


    //=================================================================//
    //文件的下载

    /**
     * 文件的下载方式一：使用copyToLocalFile
     *
     * @throws URISyntaxException
     * @throws IOException
     * @throws InterruptedException
     */
    public void downloadFile1() throws URISyntaxException, IOException {
        //1:获取FileSystem
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://hadoop10:8020"),
                new Configuration());
        //2:调用方法，实现文件的下载
        // boolean delSrc 指是否将原文件删除
        // Path src 指要下载的文件路径
        // Path dst 指将文件下载到的路径
        // boolean useRawLocalFileSystem 是否开启文件校验 就是是否生成windows系统是上
        //面那个crc文件, 设置true，不会有crc文件。设置false在本地会有crc文件。
        fileSystem.copyToLocalFile(false, new Path("/bb/cc/test1.txt"), new
                Path("D://test1_down1.txt"), false);
        //3:关闭FileSystem
        fileSystem.close();
    }

    /**
     * 文件的下载方式二：通过输入输出流
     *
     * @throws URISyntaxException
     * @throws IOException
     */
    public void downloadFile2() throws URISyntaxException, IOException {
        //1:获取FileSystem
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://hadoop10:8020"),
                new Configuration());
        //2:获取hdfs的输入流
        FSDataInputStream inputStream = fileSystem.open(new Path("/test1.txt"));
        //3:获取本地路径的输出流
        FileOutputStream outputStream = new
                FileOutputStream("D://test1_down2.txt");
        //4:文件的拷贝
        IOUtils.copy(inputStream, outputStream);
        //5:关闭流
        IOUtils.closeQuietly(inputStream);
        IOUtils.closeQuietly(outputStream);
        fileSystem.close();
    }

    //=================================================================//
    //删除文件或文件夹
    public void deleteFileOrDir() throws URISyntaxException, IOException,
            InterruptedException {
        //1:获取FileSystem（分布式文件系统）
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://hadoop10:8020"),
                new Configuration(),"root");
        //2、删除操作
        //boolean b = fileSystem.delete(new Path("/test_big.txt"));
        boolean b = fileSystem.delete(new Path("/aa"));
        System.out.println(b);
        //3、关闭FileSystem
        fileSystem.close();
    }

    //=================================================================//
    //文件的合并上传
    /**
     * 小文件的合并上传
     * @throws URISyntaxException
     * @throws IOException
     * @throws InterruptedException
     */
    public void mergeFile() throws URISyntaxException, IOException,
            InterruptedException {
        //1:获取FileSystem（分布式文件系统）
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://hadoop10:8020"),
                new Configuration(),"root");
        //2:获取hdfs大文件的输出流
        FSDataOutputStream outputStream = fileSystem.create(new
                Path("/test_big.txt"));
        //3:获取一个本地文件系统
        LocalFileSystem localFileSystem = FileSystem.getLocal(new
                Configuration());
        //4:获取本地文件夹下所有文件的详情
        FileStatus[] fileStatuses = localFileSystem.listStatus(new
                Path("D:\\input"));
        //5:遍历每个文件，获取每个文件的输入流
        for (FileStatus fileStatus : fileStatuses) {
            FSDataInputStream inputStream =
                    localFileSystem.open(fileStatus.getPath());
        //6:将小文件的数据复制到大文件
            IOUtils.copy(inputStream, outputStream);
            IOUtils.closeQuietly(inputStream);
        }
        //7:关闭流
        IOUtils.closeQuietly(outputStream);
        localFileSystem.close();
        fileSystem.close();
    }
}

打赏