第六章:HDFS的I/O流操作(扩展)
上面我们学的API操作HDFS系统都是框架封装好的。自己实现上述API的操作该怎么实现呢?可以采用IO流的方式实现数据的上传和下载。
6.1 HDFS文件上传
- 需求:把本地d盘上的QQ里面的QQWhatsnew.txt文件上传到HDFS根目录
- 编写代码
/**
* 把本地D盘上的/QQ/QQWhatsnew.txt文件上传到HDFS根目录
* @throws IOException
* @throws InterruptedException
* @throws URISyntaxException
*/
@Test
public void putFileToHDFS() throws IOException, InterruptedException, URISyntaxException {
// 1 获取文件系统
Configuration configuration = new Configuration();
FileSystem fs = FileSystem.get(new URI ("hdfs://hadoop101:9000"), configuration, "zhangyong");
// 2 创建输入流
FileInputStream fis = new FileInputStream(new File ("d:/QQ/QQWhatsnew.txt"));
// 3 获取输出流
FSDataOutputStream fos = fs.create(new Path ("/QQWhatsnew.txt"));
// 4 流对拷
IOUtils.copyBytes(fis, fos, configuration);
// 5 关闭资源
IOUtils.closeStream(fos);
IOUtils.closeStream(fis);
fs.close();
}
6.2 HDFS文件下载
- 需求:从HDFS上下载banhua.txt文件到本地e盘上
- 编写代码
/**
* 文件下载
* @throws IOException
* @throws InterruptedException
* @throws URISyntaxException
*/
@Test
public void getFileFromHDFS() throws IOException, InterruptedException, URISyntaxException{
// 1 获取文件系统
Configuration configuration = new Configuration();
FileSystem fs = FileSystem.get(new URI("hdfs://hadoop101:9000"), configuration, "zhangyong");
// 2 获取输入流
FSDataInputStream fis = fs.open(new Path("/QQWhatsnew.txt"));
// 3 获取输出流
FileOutputStream fos = new FileOutputStream(new File("d:/QQWhatsnew.txt"));
// 4 流的对拷
IOUtils.copyBytes(fis, fos, configuration);
// 5 关闭资源
IOUtils.closeStream(fos);
IOUtils.closeStream(fis);
fs.close();
}
6.3 定位文件读取
- 需求:分块读取HDFS上的大文件,比如根目录下的/hadoop- 3.1.2.tar.gz
- 编写代码
(1)下载第一块
// 1 获取文件系统
Configuration configuration = new Configuration ();
FileSystem fs = FileSystem.get (new URI ("hdfs://hadoop101:9000"), configuration, "zhangyong");
// 2 获取输入流
FSDataInputStream fis = fs.open (new Path ("/hadoop-3.1.2.tar.gz"));
// 3 创建输出流
FileOutputStream fos = new FileOutputStream (new File ("d:/hadoop-3.1.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://hadoop101:9000"), configuration, "zhangyong");
// 2 打开输入流
FSDataInputStream fis = fs.open(new Path("/hadoop-3.1.2.tar.gz"));
// 3 定位输入数据位置
fis.seek(1024*1024*128);
// 4 创建输出流
FileOutputStream fos = new FileOutputStream(new File("d:/hadoop-3.1.2.tar.gz.part2"));
// 5 流的对拷
IOUtils.copyBytes(fis, fos, configuration);
// 6 关闭资源
IOUtils.closeStream(fis);
IOUtils.closeStream(fos);
}
(3)合并文件
在Window命令窗口中进入到目录E:\,然后执行如下命令,对数据进行合并
type hadoop-3.1.2.tar.gz.part2 >> hadoop-3.1.2.tar.gz.part1
合并完成后,将hadoop-3.1.2.tar.gz.part1重新命名为hadoop-2.7.2.tar.gz。解压发现该tar包非常完整。