Java API 操作HDFS文件

使用JavaAPI操作hdfs 文件的常用操作代码

1.创建目录

​@Test
public void mkdir() throws Exception{
    fileSystem.mkdir(new Path("/hdfsapi/test"));
}

​

2.创建文件

@Test
​
public void create() throws Exception{

    FSDataOutputStream output = fileSystem.create(new Path("/hdfsapi/test/a.txt"));
    output.write("hello world".getBytes());
    output.flush();
    output.close();
}

3.重命名

@Test
​
public void rename() throws Exception{

    Path oldPath = new Path("/hdfsapi/test/a.txt");
    Path newPath = new Path("/hdfsapi/test/b.txt");
    System.out.println(fileSystem.rename(oldPath,newPath));

}

4.上传本地文件到hdfs

@Test
​
public void copyFromLocalFile() throws Exception{

    Path src = new Path("/home/hadoop/data/hello.txt");
    Path dist = new Path("/hdfsapi/test/");
    fileSystem.copyFromLocalFile(src,dist);

}

5.查看某个目录下的所有文件

@Test
​
public void listFiles() throws Exception{

    FileStatus[] listStatus = fileSystem.listStatus(new Path("/hdfsapi/test"));
    for (FileStatus fileStatus : listFiles){
        String isDir = fileStatus.isDirectory() ? "文件夹" : "文件" ;
        String permission = fileStatus.getPermission().toString();
        short replication = fileStatus.getReplication();
        long len = fileStatus.getLen();
        String path = fileStatus.getPath().toString();
        System.out.println(isDir+"\t"+permission+"\t"+replication+"\t"+long+"\t"+path)

    }
}

6.查看文件块信息

@Test
​
public void getFileBlockLocations() throws Exception{

    FileStatus fileStatus = fileSystem.getFileStatus(new Path("/hdfsapi/test/b.txt"));
    BlockLocation[] blocks = fileSystem.getFileBlockLocations(fileStatus,0,fileStatus.getLen());
    for (BlockLocation block:blocks){
        for (String host:block.getHosts()){
            System.out.println(host);
        }

    }
}

7.文件读写

public class hdfs {

    public static void main(String[] args) throws IOException {
        //请在 Begin-End 之间添加代码，完成任务要求。
        /********* Begin *********/
        Configuration conf = new Configuration();//configuration类实现hadoop各模块之间值的传递
        FileSystem fs = FileSystem.get(conf);  //获取文件系统
        Path file = new Path("/user/hadoop/myfile"); 
        FSDataOutputStream outStream = fs.create(file); //获取输出流
        outStream.writeUTF("https://www.educoder.net");//可以写入任意字符
        outStream.close();//记得关闭输出流
        FSDataInputStream inStream = fs.open(file);  //获取输入流
        String data = inStream.readUTF();  //读取文件
                /********* End *********/

    }
  }

8.文件上传

​
public class hdfs {

      /**
     * 判断路径是否存在
     */
    public static boolean test(Configuration conf, String path) throws IOException {
        /*****start*****/
        //请在此处编写判断文件是否存在的代码
        try(FileSystem fs = FileSystem.get(conf)){
            return fs.exists(new Path(path));
        } catch (IOException e){
            e.printStackTrace();
            return false;
        }   
        /*****end*****/
    }

    /**
     * 复制文件到指定路径
     * 若路径已存在，则进行覆盖
     */
    public static void copyFromLocalFile(Configuration conf, String localFilePath, String remoteFilePath) throws IOException {
        /*****start*****/
        //请在此处编写复制文件到指定路径的代码

        Path localPath = new Path(localFilePath);  
        Path remotePath = new Path(remoteFilePath);  
        try (FileSystem fs = FileSystem.get(conf)) { 
        fs.copyFromLocalFile(false, true, localPath, remotePath);  
         } catch (IOException e) {  
            e.printStackTrace();  
         }  

        /*****end*****/
    }

    /**
     * 追加文件内容
     */
    public static void appendToFile(Configuration conf, String localFilePath, String remoteFilePath) throws IOException {
       /*****start*****/
         //请在此处编写追加文件内容的代码

        Path remotePath = new Path(remoteFilePath);  
    try (FileSystem fs = FileSystem.get(conf);  
            FileInputStream in = new FileInputStream(localFilePath);) {  
        FSDataOutputStream out = fs.append(remotePath);  
        byte[] data = new byte[1024];  
        int read = -1;  
        while ((read = in.read(data)) > 0) {  
            out.write(data, 0, read);  
        }  
        out.close();  
    } catch (IOException e) {  
        e.printStackTrace();  
    }  

        /*****end*****/
    }
}

​

9.文件下载

public class hdfs {
     /**
     * 下载文件到本地
     * 判断本地路径是否已存在，若已存在，则自动进行重命名
     */
    public static void copyToLocal(Configuration conf, String remoteFilePath, String localFilePath) throws IOException {
        FileSystem fs = FileSystem.get(conf);
        Path remotePath = new Path(remoteFilePath);
        File f = new File(localFilePath);
        /*****start*****/
        /*在此添加判断文件是否存在的代码，如果文件名存在，自动重命名(在文件名后面加上 _0, _1 ...) */
        if (f.exists()) {
            System.out.println(localFilePath + " 已存在.");
            Integer i = 0;
            while (  true) {
                f = new File(  localFilePath + "_" + i.toString()     );
                if (!f.exists()  ) {
                    localFilePath = localFilePath + "_" + i.toString()      ;
                    break;
                }
            }
            System.out.println("将重新命名为: " + localFilePath);
        }
        /*****end*****/

        /*****start*****/
        // 在此添加将文件下载到本地的代码
        Path localPath = new Path(localFilePath);   
        fs.copyToLocalFile(remotePath, localPath);
       /*****end*****/
       fs.close();
    }

10.使用字节流读取数据

     /**
     * 读取文件内容
     */
    public static void cat(Configuration conf, String remoteFilePath) throws IOException {

        /*****start*****/
        //1.读取文件中的数据
        Path remotePath = new Path(remoteFilePath); 
        FileSystem fs = FileSystem.get(conf); 
        FSDataInputStream in = fs.open(remotePath); 
        BufferedReader d = new BufferedReader(new InputStreamReader(in));      
        String line = null; 
        StringBuffer buffer = new StringBuffer();
        while ((line = d.readLine()) != null) 
        { buffer.append(line); } 
        String res = buffer.toString();

        //2.将读取到的数据输出到  /tmp/output/text.txt 文件中  提示：可以使用FileWriter

        FileWriter f1=new FileWriter("/tmp/output/text.txt");
        f1.write(res);
        f1.close();
       /*****end*****/
    }

11.删除文件

     /**
     * 删除文件 
     */
    public static boolean rm(Configuration conf, String remoteFilePath) throws IOException {
        /*****start*****/        
        //请在此添加删除文件的代码
        FileSystem fs = FileSystem.get(conf);
        Path remotePath = new Path(remoteFilePath);
        boolean result = fs.delete(remotePath,false);

        return   true    ;

        /*****end*****/
    }

12.删除文件夹

public class hdfs {
        /**
     * 判断目录是否为空
     * true: 空，false: 非空
     */
    public static boolean isDirEmpty(Configuration conf, String remoteDir) throws IOException {
        FileSystem fs = FileSystem.get(conf);
        Path dirPath = new Path(remoteDir);
        RemoteIterator<LocatedFileStatus> remoteIterator = fs.listFiles(dirPath, true);
        return !remoteIterator.hasNext();
    }
    /**
     * 删除目录
     */
    public static boolean rmDir(Configuration conf, String remoteDir, boolean recursive) throws IOException {
        FileSystem fs = FileSystem.get(conf);
        Path dirPath = new Path(remoteDir);
        /* 第二个参数表示是否递归删除所有文件 */
        boolean result = fs.delete(dirPath, recursive);
        fs.close();
        return result;
    }
}

13.自定义数据输入流

public class MyFSDataInputStream extends FSDataInputStream {
    public MyFSDataInputStream(InputStream in) {
        super(in);
    }
   /**
    * 实现按行读取     * 每次读入一个字符，遇到"\n"结束，返回一行内容
     */
    public static String readline(BufferedReader br) throws IOException {
        char[] data = new char[1024];
        int read = -1;
        int off = 0; // 循环执行时，br 每次会从上一次读取结束的位置继续读取，因此该函数里，off 每次都从0开始
        while ( (read = br.read(data, off, 1)) != -1 ) {
            if (String.valueOf(data[off]).equals("\n") ) {
                off += 1;
                return String.valueOf(data, 0, read);
            }
            off += 1;
            return String.valueOf(data, 0, read);
        }
        return null;
    }
    /**
     * 读取文件内容
     */
    public static void cat(Configuration conf, String remoteFilePath) throws IOException {
        FileSystem fs = FileSystem.get(conf);
        Path remotePath = new Path(remoteFilePath);
        FSDataInputStream in = fs.open(remotePath);
        BufferedReader br = new BufferedReader(new InputStreamReader(in));
        FileWriter f = new FileWriter("/tmp/output/text.txt");
        String line = null;
        while ( (line = MyFSDataInputStream.readline(br)) != null ) {
            f.write(line);
        }
        f.close();
        br.close();
        in.close();
        fs.close();
    }
}