五、通过API操作hdfs

一、基本api操作

1、获取HDFS对象的两种方式：

方式1：

public static FileSystem initFileSystem1() throws IOException {
        //获取配置对象
        Configuration conf = new Configuration();
        //指定namenode地址
        conf.set("fs.defaultFS", "hdfs://bigdata121:9000");
        //获取hdfs文件系统访问对象
        FileSystem client = FileSystem.get(conf);
        return client;
    }

方式2：

public static FileSystem initFileSystem2() throws IOException, URISyntaxException {
        Configuration conf = new Configuration();
        //直接通过uri的方式获取hdfs文件系统访问对象
        FileSystem client = FileSystem.get(new URI("hdfs://bigdata121:9000"), conf);
        return client;
    }

后面都是通过client这个文件系统对象调用各个方法操作hdfs的。

2、configuration对象的参数值配置

//通过conf.set(key, value)的方式即可设置参数值,如
conf.set("fs.defaultFS", "hdfs://bigdata121:9000");

3、创建目录

Path p = new Path(HDFS路径);
client.mkdirs(p);

4、上传文件

```;
Path dest = new Path(HDFS路径);
Path src = new Path(本地路径);
client.copyFromLocalFile(src, dest);

//还可以设置是否删除本地文件，以及是否覆盖hdfs中的原有同名文件

5、下载文件

//用法：
client.copyToLocalFile(srcPath,destPath,)

//例子：
Path downloadFile = new Path("/king/edit_new.xml");
Path destPath = new Path("G:\edits.xml");
client.copyToLocalFile(downloadFile, destPath);
client.close();

//还可以设置是否删除hdfs中的源文件

6、删除文件或者目录

```java
/*方式1： client.delete(Path HDFS路径，boolean 是否递归删除)
         如果不是递归删除，那么在删除的目录的时候，如果目录非空，那么就会报错
*/
Path deletePath = new Path("/linux2.txt");
client.delete(deletePath, true);
client.close();

/*方式2： client.deleteOnExit(Path HDFS路径)*/
Path deletePath = new Path("/linux2.txt");
client.deleteOnExit(deletePath);       //存在裁删除
client.close();

7、查看文件属性（只能查看文件，不能查看目录）

//返回的是一个LocatedFileStatus迭代器，用法：
RemoteIterator<LocatedFileStatus> pathList = client.listFiles(HDFS路径, recursive);
recursive表示是否递归显示子目录下的内容

//例子：
public void listFileMetaData() throws Exception{
        FileSystem client = initFileSystem2();
        Path listPath = new Path("/");
        //获取指定路径下列表，不递归显示，返回一个迭代器
        RemoteIterator<LocatedFileStatus> pathList = client.listFiles(listPath, false);
        while (pathList.hasNext()) {
            LocatedFileStatus i = pathList.next();
            //获取文件名
            System.out.println(i.getPath().getName());
            //文件权限
            System.out.println(i.getPermission());
            //文件属主
            System.out.println(i.getOwner());
            //文件数组
            System.out.println(i.getGroup());
            //文件大小，单位Byte
            System.out.println(i.getLen());
            //文件的block的大小
            System.out.println("blocksize:" + i.getBlockSize());

            //获取文件的block地址
            BlockLocation[] bl = i.getBlockLocations();
            for (BlockLocation b:bl) {
                //获取每个block 的偏移地址
                System.out.println("offset:" + b.getOffset());
                //获取当前副本的block所在的所有datanode的主机名
                String[] hosts = b.getHosts();
                for (String h:hosts) {
                    System.out.println(h);
                }
            }
            System.out.println("===========================");
        }

8、查看文件和目录的属性

//返回的是一个FileStatus数组，无法递归显示子目录下的内容，但是能查看子目录本身的属性,用法
FileStatus[] f = client.listStatus(Path);

public void judgeFile() throws Exception{
        FileSystem client = initFileSystem2();
        Path path = new Path("/");
        //获取FileSstatus对象
        FileStatus[] fileStatuses = client.listStatus(path);
        //通过Filestatus对象获取文件或者目录的属性
        for (FileStatus f:fileStatuses) {
            System.out.println("文件名:" + f.getPath().getName());
            System.out.println("权限:" + f.getPermission());
            System.out.println("大小:" + f.getLen());
            System.out.println("==========================");
        }
        client.close();
    }

9、文件类型判断

//通过上面的FileStatus和LocatedFileStatus 都可以调用内部的方法判断当前是文件还是目录
FileStatus对象.isFile()
LocatedFileStatus对象.isFile()

FileStatus对象.isDirectory()
LocatedFileStatus对象.isDirectory()

二、以IO流操作hdfs

1、以IO流方式上传文件
主要用到的是下面两个方法：

//这是HDFS专用的数据流输出流
FSDataOutputStream fos = client.create(Path)

//create方法还有以下参数：
boolean overwrite：如果文件文件已存在，是否覆盖，默认是true
short replication：可以指定副本数，不指定就以hdfs的配置为准
int bufferSize：缓冲区大小
long blockSize：指定自己使用的块大小
FsPermission var2：指定权限
ChecksumOpt checksumOpt：指定校验值

//下面是对接输入流和输出流的工具
IOutils.copyBytes(inputstream,outputstream,buffsize,close)

inputstream   输入流
outputstream  输出流
buffsize      缓冲区
close         是否关闭流

例子：

 @Test
    public void putFileFromIO() throws Exception {
        FileSystem client = initFileSystem2();

        //创建本地文件字节输入流
        InputStream fis = new FileInputStream("E:\\file\\big data\\java se\\第十八章 Java文件与IO流.md");

        //创建hdfs文件字节输出流，注意，创建输出流文件的时候，一定要指定文件名，否则会报错
        Path uploadPath = new Path("/第十八章 Java文件与IO流.md");
        FSDataOutputStream fos = client.create(uploadPath);

        //输入流和输出流对接
        try {
            //输入流和输出流拷贝，后面false表示不关闭流
            IOUtils.copyBytes(fis, fos, 1024，false);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            //关闭输入流和输出流
            IOUtils.closeStream(fis);
            IOUtils.closeStream(fos);
        }
    }

2、以IO流方式下载文件

Path getPath = new Path("/dog.txt");
FSDataInputStream fis = client.open(getPath);

例子：

/*
    * 写入有两种方式：
    * 1、使用流的read/write方法
    * 2、使用IOUtils.copyBytes(instream,outstream,buffize)这个工具
    * */
    @Test
    public void getFileFromIO() throws Exception{
        FileSystem client = initFileSystem2();

        //创建本地输出流，保存内容
        OutputStream fos = new FileOutputStream("F:\\edit_new.xml");

        //创建hdfs输入流
        Path getPath = new Path("/dog.txt");
        FSDataInputStream fis = client.open(getPath);

        try {
            IOUtils.copyBytes(fis, System.out, 1024);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            IOUtils.closeStream(fis);
            IOUtils.closeStream(fos);
        }

    }