HDFS 的 I/O 流操作

水深无声 2022-02-25 12:58 194阅读 0赞

### HDFS 文件上传 ###

@Test
    public void putFileToHDFS() throws Exception { 
    
        // 1 获取文件系统
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "root");
    
        // 2 创建输入流
        FileInputStream fis = new FileInputStream(new File("/Users/ylj/demo/io.txt"));
        // 3 获取输出流
        FSDataOutputStream fos = fs.create(new Path("/demo/test/io.txt"));
        // 4 流对拷
        IOUtils.copyBytes(fis, fos, configuration);
        // 5 关闭资源
        IOUtils.closeStream(fis);
        IOUtils.closeStream(fos);
    
        fs.close();
        
        System.out.println("~~ok~~");
    }

### HDFS 文件下载 ###

@Test
    public void getFileFromHDFS() throws Exception{ 
        // 1 获取文件系统
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "root");
        // 2 获取输人流
        FSDataInputStream fis = fs.open(new Path("/demo/test/io.txt"));
        // 3 获取输出流
        FileOutputStream fos = new FileOutputStream(new File("/Users/ylj/demo/io_bak.txt"));
        // 4 流的对拷
        IOUtils.copyBytes(fis, fos, configuration);
        // 5 关闭资源
        IOUtils.closeStream(fis);
        IOUtils.closeStream(fos);
    
        fs.close();
    
        System.out.println("~~ok~~");
    }

### 定位文件读取 ###

@Test
    public void readFileSeek1() throws Exception { 
        // 1 获取文件系统
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "root");
        // 2 获取输入流
        FSDataInputStream fis = fs.open(new Path("/test/demo/hadoop-2.7.2.tar.gz"));
        // 3 创建输出流
        FileOutputStream fos = new FileOutputStream(new File("/Users/ylj/demo/hadoop-2.7.2.tar.gz.part1"));
    
        // 4 流的拷贝
        byte[] buf = new byte[1024];
        for (int i = 0; i < 1024 * 128; i++) { 
            fis.read(buf);
            fos.write(buf);
        }
        //5 关闭资源
        IOUtils.closeStream(fis);
        IOUtils.closeStream(fos);
    
        fs.close();
    
        System.out.println("~~ok~~");
    }
    
    @Test
    public void readFileSeek2() throws Exception { 
        // 1 获取文件系统
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration,"root");
        // 2 打开输入流
        FSDataInputStream fis = fs.open(new Path("/test/demo/hadoop-2.7.2.tar.gz"));
        // 3 定位输入数据位置
        fis.seek(1024 * 1024 * 128);
        // 4 创建输出流
        FileOutputStream fos = new FileOutputStream(new File("/Users/ylj/demo/hadoop-2.7.2.tar.gz.part2"));
        // 5 流的对拷
        IOUtils.copyBytes(fis, fos, configuration);
        // 6 关闭资源
        IOUtils.closeStream(fis);
        IOUtils.closeStream(fos);
    
        fs.close();
    
        System.out.println("~~ok~~");
    }

文件合并

cat hadoop-2.7.2.tar.gz.part2 >> hadoop-2.7.2.tar.gz.part1