io 流操作hdfs

Posted 2021-01-24 lijins

hdfs 文件上传

本地   -------->    文件系统对象   -------->    hdfs 文件系统

           输入流                               输出流  

// 将流从本地 上传到  hdfs 文件系统。
    @Test
    public void ioPut() throws IOException, InterruptedException, URISyntaxException{
        //1  获取文件系统对象
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://192.168.59.11:9000"), conf, "ljs");
        
        //2  输出流   流到 hdfs 
        FSDataOutputStream fo = fs.create(new Path("/user/ljs/job"));
        
        //3 定义输出流  来自 本地磁盘  
        InputStream is = new FileInputStream(new File("d:/job.txt"));
        
        //4 流对接 
        IOUtils.copyBytes(is, fo, 100);
        
        IOUtils.closeStream(fs);
        //5 关闭流
    }

 

 

hdfs 文件下载

本地 <-----------  文件系统对象  < --------------------  hdfs文件系统 集群

           输出流                                     输出流

@Test
    public void ioGet() throws IOException, InterruptedException, URISyntaxException{
        
        // 1 创建文件系统
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://192.168.59.11:9000"),
                conf, "ljs");
        // 2 创建 输入流  来着 hdfs 文件系统
        FSDataInputStream fis = fs.open(new Path("/user/ljs/cook.txt"));
        // 3 创建输出流   送到 本地磁盘。
        OutputStream os = new FileOutputStream(new File("d:/cook.txt"));
        // 4  流对接
        IOUtils.copyBytes(fis, os, 20);
        // 5 关闭流
        IOUtils.closeStream(fis);
     }

 

 

定位文件读取

 

读取第一块   128M  

    
    // 从hdfs文件系统中获取第一块block (128M)
    @Test
    public void fileSeek1() throws IOException, InterruptedException, URISyntaxException{
        // 1  获取文件系统对象
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://192.168.59.11:9000"), conf, "ljs");
        // 2 创建打开输入流  
        FSDataInputStream fis = fs.open(new Path("/user/ljs/hadoop-2.7.2.tar.gz"));
        // 3 创建 输出流 
        OutputStream fos = new FileOutputStream(new File("d:/hadoop-2.7.2.tar.gz.part01"));
        // 4 流对接
        byte [] buff = new byte[1024]; // 1kB;
        for(int i =0; i < 1024*128; ++i){
            fis.read(buff);
            fos.write(buff);    
        }
        // 5 关闭流
        fis.close();
        fos.close();
    }

 

 

从第二块文件开始读取

@Test
	public void fileSeek02() throws IOException, InterruptedException, URISyntaxException{
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.get(new URI("hdfs://192.168.59.11:9000"), conf, "ljs");
		FSDataInputStream fis = fs.open(new Path("/user/ljs/hadoop-2.7.2.tar.gz"));
		OutputStream fos = new FileOutputStream(new File("d:/hadoop-2.7.2.tar.gz.part02"));
		
		fis.seek(1024*1024*128);
		IOUtils.copyBytes(fis, fos,1024);
		
		fis.close();
		fos.close();
		
	}