HDFS中JAVA API的使用

傷城~ 2022-06-10 04:52 178阅读 0赞

HDFS是一个分布式文件系统，既然是文件系统，就可以对其文件进行操作，比如说新建文件、删除文件、读取文件内容等操作。下面记录一下使用JAVA API对HDFS中的文件进行操作的过程。

对分HDFS中的文件操作主要涉及一下几个类：

Configuration类：该类的对象封转了客户端或者服务器的配置。

FileSystem类：该类的对象是一个文件系统对象，可以用该对象的一些方法来对文件进行操作。FileSystem fs = FileSystem.get(conf);通过FileSystem的静态方法get获得该对象。

FSDataInputStream和FSDataOutputStream：这两个类是HDFS中的输入输出流。分别通过FileSystem的open方法和create方法获得。

具体如何对文件操作清下下面例子：

package com.hadoop.hdfs;
    
    import java.io.FileOutputStream;
    import java.io.IOException;
    import java.io.OutputStream;
    import java.net.URI;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.FSDataInputStream;
    import org.apache.hadoop.fs.FileStatus;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IOUtils;
    
    public class Demo { 
    
        static String HDFS = "hdfs://192.168.7.12:9000";
        Configuration conf = new Configuration();
    
        public static void main(String[] args) throws IOException {
    
            Demo dm = new Demo();
            //dm.mkdir();
        }
    
        //创建新文件
        public void createFile() throws IOException{
            FileSystem fs = FileSystem.get(conf);
            Path dstPath = new Path(HDFS+"/test01"); //目标路径
            //打开一个输出流
            FSDataOutputStream outputStream = fs.create(dstPath);
            byte[] contents ={h,e,l,l,o};
            outputStream.write(contents);
            outputStream.close();
            fs.close();
            System.out.println("文件创建成功！");
        }
        //创建文件目录
        public void mkdir() throws IOException{
            FileSystem fs = FileSystem.get(URI.create(HDFS), conf);
            fs.mkdirs(new Path(HDFS+"/test01"));
            fs.close();
        }
        //删除一个文件或目录
        public void rm() throws IOException{
            FileSystem fs =FileSystem.get(URI.create(HDFS), conf);
            fs.delete(new Path(HDFS+"/test01"), true);
            fs.close();
        }
        //查看文件目录
        public void lookup() throws IOException{
            FileSystem fs =FileSystem.get(URI.create(HDFS), conf);
            FileStatus[] fss = fs.listStatus(new Path(HDFS+"/test01"));
            for(FileStatus f: fss){
                System.out.println(f.getPath().getName());
            }
            fs.close();
        }
        //本地上传一个文件
        public void upload() throws IOException{
            FileSystem fs =FileSystem.get(URI.create(HDFS), conf);
            //调用文件系统的文件复制函数,前面参数是指是否删除原文件，true为删除，默认为false
            fs.copyFromLocalFile(false,new Path("D:/java/a.txt"), new Path(HDFS+"/test/"));
            fs.close();
        }
        //将一个文件下载到本地
        public void dowmload() throws IOException{
            FileSystem fs =FileSystem.get(URI.create(HDFS), conf);
            FSDataInputStream fsd = fs.open(new Path(HDFS+"/test/"));
            OutputStream out = new FileOutputStream("D:/java/a.txt");
            //输入流、输出流、缓存大小、表明拷贝完成后我们关闭拷贝源可拷贝目的地
            IOUtils.copyBytes(fsd, out, 2048, true);
            fs.close();
        }
        //文件重命名
        public static void rename(String oldName,String newName) throws IOException{
            Configuration conf = new Configuration();
            FileSystem fs = FileSystem.get(conf);
            Path oldPath = new Path(oldName);
            Path newPath = new Path(newName);
            boolean isok = fs.rename(oldPath, newPath);
            if(isok){
                System.out.println("rename ok!");
            }else{
                System.out.println("rename failure");
            }
            fs.close();
        }
    
    }