HDFS系列(7) | HDFS的 I/O流 API操作
上一篇博文我们分享的API操作HDFS系统都是框架封装好的。那么如果我们想自己实现上述API的操作该怎么实现呢？ 此次物品们介绍的是采用I/O流的方式实现数据的上传和下载。
一. HDFS文件上传
需求：把本地d盘上的lisen.txt文件上传到HDFS根目录
- 1. 代码
package com.buwenbuhuo.hdfs;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.junit.Test;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
/**
* @author buwenbuhuo
* @create 2020-04-22 18:26
* com.buwenbuhuo.hdfs - the name of the target package where the new class or interface will be created.
* hdfs0422 - the name of the current project.
*/
public class HDFSClient {
@Test
public void putFileToHDFS() throws IOException, InterruptedException, URISyntaxException {
// 1 获取文件系统
Configuration configuration = new Configuration();
FileSystem fs = FileSystem.get(URI.create("hdfs://hadoop001:9000"), configuration, "bigdata");
// 2 创建输入流
FileInputStream fis = new FileInputStream(new File("d:/lisen.txt"));
// 3 获取输出流
FSDataOutputStream fos = fs.create(new Path("/lisen.txt"));
// 4 流对拷
IOUtils.copyBytes(fis, fos, configuration);
// 5 关闭资源
IOUtils.closeStream(fos);
IOUtils.closeStream(fis);
fs.close();
}
}- 2. 结果
1
二. HDFS文件下载
需求：从HDFS上下载lisen.txt文件到本地d盘上
- 1.代码
// 文件下载
@Test
public void getFileFromHDFS() throws IOException, InterruptedException, URISyntaxException{
// 1 获取文件系统
Configuration configuration = new Configuration();
FileSystem fs = FileSystem.get(URI.create("hdfs://hadoop001:9000"), configuration, "bigdata");
// 2 获取输入流
FSDataInputStream fis = fs.open(new Path("/lisen.txt"));
// 3 获取输出流
FileOutputStream fos = new FileOutputStream(new File("d:/lisen1.txt"));
// 4 流的对拷
IOUtils.copyBytes(fis, fos, configuration);
// 5 关闭资源
IOUtils.closeStream(fos);
IOUtils.closeStream(fis);
fs.close();
}- 2. 结果
2
三. 定位文件读取
需求：分块读取HDFS上的大文件，比如根目录下的/hadoop-2.7.2.tar.gz
- 1. 下载第一块
@Test
public void readFileSeek1() throws IOException, InterruptedException, URISyntaxException{
// 1 获取文件系统
Configuration configuration = new Configuration();
FileSystem fs = FileSystem.get(URI.create("hdfs://hadoop001:9000"), configuration, "bigdata");
// 2 获取输入流
FSDataInputStream fis = fs.open(new Path("/hadoop-2.7.2.tar.gz"));
// 3 创建输出流
FileOutputStream fos = new FileOutputStream(new File("d:/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(URI.create("hdfs://hadoop001:9000"), configuration, "bigdata");
// 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("d:/hadoop-2.7.2.tar.gz.part2"));
// 5 流的对拷
IOUtils.copyBytes(fis, fos, configuration);
// 6 关闭资源
IOUtils.closeStream(fis);
IOUtils.closeStream(fos);
}- 3. 合并文件 在Window命令窗口中进入到目录D:\，然后执行如下命令，对数据进行合并
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包非常完整。
本文参与 腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。
原始发表：2020/04/25 ，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读
腾讯云开发者
