HDFS FileSystem(一) 创建对象

 

1.非HA (namenode没有做高可用)

直接在URI中写明hdfs地址即可。如：

    static FileSystem fs;

    static {
        try {
            fs = FileSystem.get(new URI("hdfs://cluster-host1:9000"),new Configuration(),"hadoop");
        } catch (IOException e) {
            e.printStackTrace();
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }
    }
    @Test
    public void mkdir() throws IOException {
        String path = "/test/fs";
        fs.mkdirs(new Path(path),new FsPermission("755"));//绝对路径，相对路径会在每个用户下
    }
    public static void listFiles(String dirName)throws IOException {
        Path f = new Path(dirName);
        FileStatus[] status =fs.listStatus(f);
        System.out.println(dirName +" has all files:");
        for (int i = 0; i<status.length; i++) {
            System.out.println(status[i].getPath().toString());
            System.out.print("  | 是否目录："+status[i].isDirectory());
            System.out.print("  | 是否文件："+status[i].isFile());
            System.out.print("  | permission:"+status[i].getPermission());
            System.out.print("  | owner:"+status[i].getOwner());
            System.out.println();
        }
    }

2.HA

在使用Hadoop Java API访问HDFS集群时，在创建FileSystem对象时，直接指定NameNode的IP以及端口号即可。但是在HA模式下，访问HDFS集群却有一些不同，需要指定NameSpace和主备NameNode的IP以及端口等信息，具体操作方式见如下代码：

方式一：

public class FileSystemHA {

    //方式一：这种方式测试通过
    @Test
    public void test1() throws Exception{
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS","hdfs://hadoop-ns1");//nameservices地址
        conf.set("dfs.nameservices", "hadoop-ns1");
        conf.set("dfs.ha.namenodes.hadoop-ns1", "nn1,nn2");
        conf.set("dfs.namenode.rpc-address.hadoop-ns1.nn1", "hadoop-master1:8020");
        conf.set("dfs.namenode.rpc-address.hadoop-ns1.nn2", "hadoop-master2:8020");
        conf.set("dfs.client.failover.proxy.provider.ns1", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");
        FileSystem fs = FileSystem.get(URI.create("hdfs://hadoop-ns1"), conf, "hadoop");

        String dirName = "/";
        Path f = new Path(dirName);
        FileStatus[] status =fs.listStatus(f);
        System.out.println(dirName +" has all files:");
        for (int i = 0; i<status.length; i++) {
            System.out.println(status[i].getPath().toString());
            System.out.print("  | 是否目录："+status[i].isDirectory());
            System.out.print("  | 是否文件："+status[i].isFile());
            System.out.print("  | permission:"+status[i].getPermission());
            System.out.print("  | owner:"+status[i].getOwner());
            System.out.println();
        }
    }
}

 方式二：配置文件从hadoop-master1复制的

import com.google.common.base.Preconditions;
import com.ultiwill.common.constant.CommonConstant;
import java.io.File;
import java.io.IOException;
import java.net.MalformedURLException;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class HdfsUtil {
    private static final Logger log = LoggerFactory.getLogger(HdfsUtil.class);
    private static FileSystem fs;

    private HdfsUtil() {
    }

    public static FileSystem getSingletonFs() {
        if (fs == null) {
            Class var0 = HdfsUtil.class;
            synchronized(HdfsUtil.class) {
                if (fs == null) {
                    fs = getFs();
                }
            }
        }

        return fs;
    }

    public static FileSystem getFs() {
        FileSystem fs = null;
        log.info("准备初始化hdfs连接");

        try {
            fs = FileSystem.get(getConf());
        } catch (IOException var2) {
            log.error("连接hdfs异常", var2);
        }

        log.info("初始化hdfs连接成功");
        Preconditions.checkNotNull(fs, "创建hdfs连接失败");
        return fs;
    }

    public static Configuration getConf() {
        String defaultHdfsConfDir = CommonConstant.HADOOP_CONF_PATH;
        String hdfsConfEnv = System.getenv("HADOOP_CONF_DIR");   //文件放在环境变量下
        if (StringUtils.isNotBlank(hdfsConfEnv)) {
            defaultHdfsConfDir = hdfsConfEnv;
        }

        log.info("读取hdfs配置文件目录：{}", defaultHdfsConfDir);
        File hdfsSiteFile = new File(defaultHdfsConfDir + File.separator + "hdfs-site.xml");
        File coreSiteFile = new File(defaultHdfsConfDir + File.separator + "core-site.xml");
        Preconditions.checkArgument(hdfsSiteFile.exists(), "hdfs-site.xml文件缺失");
        Preconditions.checkArgument(coreSiteFile.exists(), "core-site.xml文件缺失");
        Configuration conf = new Configuration();

        try {
            conf.addResource(hdfsSiteFile.toURI().toURL());
            conf.addResource(coreSiteFile.toURI().toURL());
            conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
        } catch (MalformedURLException var6) {
            log.error("hdfs配置文件加载异常", var6);
        }

        return conf;
    }
}