尚硅谷大数据技术之Hadoop(HDFS)第3章 HDFS客户端操作(开发重点)

3.3 HDFS的I/O流操作

上面我们学的API操作HDFS系统都是框架封装好的。那么如果我们想自己实现上述API的操作该怎么实现呢?

我们可以采用IO流的方式实现数据的上传和下载。

3.3.1 HDFS文件上传

1.需求:把本地e盘上的banhua.txt文件上传到HDFS根目录

2.编写代码

@Test

public void putFileToHDFS() throws IOException, InterruptedException, URISyntaxException {

// 1 获取文件系统

Configuration configuration = new Configuration();

FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "atguigu");

// 2 创建输入流

FileInputStream fis = new FileInputStream(new File("e:/banhua.txt"));

// 3 获取输出流

FSDataOutputStream fos = fs.create(new Path("/banhua.txt"));

// 4 流对拷

IOUtils.copyBytes(fis, fos, configuration);

// 5 关闭资源

IOUtils.closeStream(fos);

IOUtils.closeStream(fis);

    fs.close();

}

3.3.2 HDFS文件下载

1.需求:从HDFS上下载banhua.txt文件到本地e盘上

2.编写代码

// 文件下载

@Test

public void getFileFromHDFS() throws IOException, InterruptedException, URISyntaxException{

// 1 获取文件系统

Configuration configuration = new Configuration();

FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "atguigu");

// 2 获取输入流

FSDataInputStream fis = fs.open(new Path("/banhua.txt"));

// 3 获取输出流

FileOutputStream fos = new FileOutputStream(new File("e:/banhua.txt"));

// 4 流的对拷

IOUtils.copyBytes(fis, fos, configuration);

// 5 关闭资源

IOUtils.closeStream(fos);

IOUtils.closeStream(fis);

fs.close();

}

3.3.3 定位文件读取

1.需求:分块读取HDFS上的大文件,比如根目录下的/hadoop-2.7.2.tar.gz

2.编写代码

(1)下载第一块

@Test

public void readFileSeek1() throws IOException, InterruptedException, URISyntaxException{

// 1 获取文件系统

Configuration configuration = new Configuration();

FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "atguigu");

// 2 获取输入流

FSDataInputStream fis = fs.open(new Path("/hadoop-2.7.2.tar.gz"));

// 3 创建输出流

FileOutputStream fos = new FileOutputStream(new File("e:/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();

}

(2)下载第二块

@Test

public void readFileSeek2() throws IOException, InterruptedException, URISyntaxException{

// 1 获取文件系统

Configuration configuration = new Configuration();

FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "atguigu");

// 2 打开输入流

FSDataInputStream fis = fs.open(new Path("/hadoop-2.7.2.tar.gz"));

// 3 定位输入数据位置

fis.seek(1024*1024*128);

// 4 创建输出流

FileOutputStream fos = new FileOutputStream(new File("e:/hadoop-2.7.2.tar.gz.part2"));

// 5 流的对拷

IOUtils.copyBytes(fis, fos, configuration);

// 6 关闭资源

IOUtils.closeStream(fis);

IOUtils.closeStream(fos);

}

(3)合并文件

在Window命令窗口中进入到目录E:\,然后执行如下命令,对数据进行合并

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

合并完成后,将hadoop-2.7.2.tar.gz.part1重新命名为hadoop-2.7.2.tar.gz。解压发现该tar包非常完整。