hadoop学习03——JavaAPI操作HDFS

一、HDFS的java操作

• hdfs在生产应用中主要是客户端的开发，其核心步骤是从hdfs提供的api中构造一个hdfs的访问客户端对象，然后通过该客户端对象操作（增删改查）HDFS上的文件

1.1 获取api中的客户端对象

• 在java中操作hdfs，首先要获得一个客户端实例

Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(conf);

• 我们的操作目标是HDFS，所以获取到的fs对象应该是DistributeFileSystem的实例：get方法从conf中的一个参数fs.defaultFS的配置判断实例化哪种客户端类，如果我们的代码中没有指定fs.defaultFS，并且工程classpath下也没有给定相应的配置，conf中的默认值就来自于hadoop的jar包中的core-default.xml，默认值为：file:///，去的将不是一个DistributedFiledSystem的实例，而是一个本地文件系统的客户端对象。

1.2 DistributedFileSystem实例对象所具备的方法

1.3 HDFS客户端操作数据代码实例

1.3.1 文件的增删改查

package hadoop.hdfs;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Iterator;
import java.util.Map.Entry;

/*
* 客户端去操作hdfs时候，是有一个用户身份的
* 默认情况下，hdfs客户端api会从jvm中获取一个参数来作为自己的用户身份，-HADOOP_USER_NAME=hadoop
* 也可以在构造fs的时候作为参数传进去
* */

public class HdfsClientDemo {

    FileSystem fs = null;
    Configuration conf = null;
    @Before
    public void init() throws IOException, URISyntaxException, InterruptedException {
        conf = new Configuration();

        // 拿到一个文件系统操作的客户端实例对象，可以直接传入URI和用户身份
        fs = FileSystem.get(new URI("hdfs://master:9000"), conf, "root");
    }

    // 上传文件
    @Test
    public void testUpload() throws IOException, InterruptedException {
        Thread.sleep(5000);
        fs.copyFromLocalFile(new Path("/Users/shiwenhu/Downloads/readme"), new Path("/readme.copy"));
        fs.close();
    }

    // 打印参数
    @Test
    public void testConf() {
        Iterator<Entry<String, String>> it = conf.iterator();
        while(it.hasNext()) {
            Entry<String, String> ent = it.next();
            System.out.println(ent.getKey() + ">>>" + ent.getValue());
        }
    }

    // 下载文件
    @Test
    public void testDownload() throws IOException {
        fs.copyToLocalFile(new Path("/readme.copy"), new Path("/Users/shiwenhu/Desktop/"));
        fs.close();
    }

    @Test
    public void testMkdir() throws IOException {
        boolean mkdirs = fs.mkdirs(new Path("/testmkdir/aaa/bbb"));
        System.out.println(mkdirs);
    }

    @Test
    public void testDelete() throws IOException {
        boolean flag = fs.delete(new Path("/testmkdir/aaa"), true);
        System.out.println(flag);
    }

    /*
    * 递归列出指定目录下所有子文件夹中的文件
    * */
    @Test
    public void testLs() throws IOException {
        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);
        while(listFiles.hasNext()) {
            LocatedFileStatus fileStatus = listFiles.next();
            System.out.println("blocksize:" + fileStatus.getBlockSize());
            System.out.println("owner:" +fileStatus.getOwner());
            System.out.println("Replication:" + fileStatus.getReplication());
            System.out.println("Permission:" + fileStatus.getPermission());
            System.out.println("Name:" + fileStatus.getPath().getName());
            System.out.println("------------------------------------------");
            BlockLocation[] blockLocations = fileStatus.getBlockLocations();
            for (BlockLocation b : blockLocations) {
                System.out.println("块起始偏移量：" + b.getOffset());
                System.out.println("块长度：" + b.getLength());
                System.out.println("块的名字：" + b.getNames());
            }
        }
    }

    @Test
    public void testLs2() throws IOException {
        FileStatus[] fileStatuses = fs.listStatus(new Path("/"));
        for (FileStatus file : fileStatuses) {
            System.out.println("name" + file.getPath().getName());
            System.out.println(file.isFile()?"file":"directory");
        }
    }

}

1.3.2 通过流的方式访问hdfs

package hadoop.hdfs;

/*
* 用流的方式来操作hdfs上的文件
* 可以实现读取指定偏移量范围的数据
*
* */

import org.apache.commons.io.IOUtils;
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.junit.Before;
import org.junit.Test;

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

public class HdfsSteamAccess {
    FileSystem fs = null;
    Configuration conf = null;

    @Before
    public void init() throws URISyntaxException, IOException, InterruptedException {
        conf = new Configuration();

        // 可以直接传入uri和用户身份
        fs = FileSystem.get(new URI("hdfs://master:9000"), conf, "root");

    }

    /*
    * 通过流的方式上传文件到hdfs
    * */
    @Test
    public void testWrite() throws IOException {
        FSDataOutputStream outputStream = fs.create(new Path("/openGauss-1.0.0-openEuler-64bit.tar.gz"), true);
        FileInputStream fileInputStream = new FileInputStream("/Users/shiwenhu/Downloads/openGauss-1.0.0-openEuler-64bit.tar.gz");
        IOUtils.copy(fileInputStream, outputStream);
    }

    /*
    * 通过流的方式下载文件到本地
    * */
    @Test
    public void testDownLoad() throws IOException {
        FSDataInputStream inputStream = fs.open(new Path("/openGauss-1.0.0-openEuler-64bit.tar.gz"));
        FileOutputStream outputStream = new FileOutputStream("/Users/shiwenhu/Desktop/output");
        IOUtils.copy(inputStream, outputStream);
    }

    /*
    * 随机读取
    * */
    @Test
    public void testRandomAccess() throws IOException {
        FSDataInputStream inputStream = fs.open(new Path("/openGauss-1.0.0-openEuler-64bit.tar.gz"));
        inputStream.seek(12);
        FileOutputStream outputStream = new FileOutputStream("/Users/shiwenhu/Desktop/out.part1");
        IOUtils.copy(inputStream, outputStream);
    }

    /*
    * 显示文件内容到屏幕上
    * */
    @Test
    public void testCat() throws IOException {
        FSDataInputStream inputStream = fs.open(new Path("/openGauss-1.0.0-openEuler-64bit.tar.gz"));
        IOUtils.copy(inputStream, System.out);
    }


}