HDFS 客户端操作
1.HDFS客户端环境准备
（1）根据自己的电脑的操作系统拷贝对应的编译后的hadoop jar包到非中文路径（例如我的MAC：/Users/luomingkui/hadoop/hadoop-2.7.2）。
（2）配置MAC环境变量
（3）创建一个Maven工程HdfsClientDemo
（4）导入相应的依赖
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>RELEASE</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-core</artifactId>
<version>2.8.2</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>2.7.2</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>2.7.2</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-hdfs</artifactId>
<version>2.7.2</version>
</dependency>
</dependencies>
（5）创建包名：com.luomk.hdfs
（6）创建HdfsClient类
public class HdfsClient {
@Test
public void testMkdirs() throws IOException, InterruptedException, URISyntaxException {
// 1 获取文件系统
Configuration configuration = new Configuration();
// 配置在集群上运行
// configuration.set("fs.defaultFS", "hdfs://hadoop103:9000");
// FileSystem fs = FileSystem.get(configuration);
FileSystem fs = FileSystem.get(new URI("hdfs://hadoop103:9000"), configuration, "luomk");
// 2 创建目录
fs.mkdirs(new Path("/0115/daxian/luomk"));
// 3 关闭资源
fs.close();
}
}
（7）执行程序
运行时需要配置用户名称
客户端去操作HDFS时，是有一个用户身份的默认情况下，HDFS客户端API会从JVM中获取一个参数来作为自己的用户身份：-DHADOOP_USER_NAME = luomk，luomk为用户名称。
（8）注意：如果eclipse打印不出日志，在控制台上只显示
1.log4j：WARN记录器（org.apache.hadoop.util.Shell）找不到appender。
2.log4j：WARN请正确初始化log4j系统。
3.log4j：WARN请参阅http://logging.apache.org/log4j/1.2/faq.html#noconfig了解更多信息。
需要在项目的的src / main /资源目录下，新建一个文件，命名为“log4j.properties”，在文件中填入
log4j.rootLogger =信息，标准输出
log4j.appender.stdout = org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern =％d％p [％c] - ％m％n
log4j.appender.logfile = org.apache.log4j.FileAppender
log4j.appender.logfile.File =目标/ spring.log
log4j.appender.logfile.layout = org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern =％d％p [％c] - ％m％n
2.HDFS的API操作
（1）HDFS文件上传（测试参数优先级）
（1）编写源代码
@Test
public void testPut() throws Exception {
//获取配置文件
Configuration configuration = new Configuration();
configuration.set("dfs.replication", "2");
//获取文件系统
//上传文件
fs.copyFromLocalFile(new Path("/Users/luomingkui/Downloads/wc.input"), new Path("/luomk/wc.input"));
System.out.println("Over!");
//关闭资源
fs.close();
}
（2）将hdfs-site.xml拷贝到项目的根目录下
<?xml version="1.0" encoding="UTF-8"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
<property>
<name>dfs.replication</name>
<value>1</value>
</property>
</configuration>
（3）参数优先级
参数优先级排序：（1）客户端代码中设置的值>（2）classpath下的用户自定义配置文件>（3）然后是服务器的默认配置
（2）HDFS文件下载
@Test
public void testCopyToLocalFile() throws IOException, InterruptedException, URISyntaxException{
// 1 获取文件系统
Configuration configuration = new Configuration();
// 2 执行下载操作
// boolean delSrc 指是否将原文件删除
// Path src 指要下载的文件路径
// Path dst 指将文件下载到的路径
// boolean useRawLocalFileSystem 是否开启文件校验
fs.copyToLocalFile(false,new Path("/luomk/wc.input"), new Path("/Users/luomingkui/Downloads/"));
// 3 关闭资源
fs.close();
}
（3）HDFS文件夹删除
@Test
public void testDelete() throws IOException, InterruptedException, URISyntaxException{
// 1 获取文件系统
Configuration configuration = new Configuration();
FileSystem fs = FileSystem.get(new URI("hdfs://hadoop103:9000"), configuration, "luomk");
// 2 执行删除
fs.delete(new Path("/over"), true);
// 3 关闭资源
fs.close();
}
（4）HDFS文件名更改
@Test
public void testRename() throws IOException, InterruptedException, URISyntaxException{
// 1 获取文件系统
Configuration configuration = new Configuration();
FileSystem fs = FileSystem.get(new URI("hdfs://hadoop103:9000"), configuration, "luomk");
// 2 修改文件名称
fs.rename(new Path("/luomk/wc.input"), new Path("/luomk/wc1.input"));
// 3 关闭资源
fs.close();
}
(5) HDFS文件详情查看
查看文件名称、权限、长度、块信息
@Test
public void testListFiles() throws IOException, InterruptedException, URISyntaxException{
// 1获取文件系统
Configuration configuration = new Configuration();
// 2 获取文件详情
RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);
while(listFiles.hasNext()){
LocatedFileStatus status = listFiles.next();
// 输出详情
// 文件名称
System.out.println(status.getPath().getName());
// 长度
System.out.println(status.getLen());
// 权限
System.out.println(status.getPermission());
// z组
System.out.println(status.getGroup());
// 获取存储的块信息
BlockLocation[] blockLocations = status.getBlockLocations();
for (BlockLocation blockLocation : blockLocations) {
// 获取块存储的主机节点
String[] hosts = blockLocation.getHosts();
for (String host : hosts) {
System.out.println(host);
}
}
System.out.println("----------------分割线-----------");
}
}
(6) HDFS文件和文件夹判断
@Test
public void testListStatus() throws IOException, InterruptedException, URISyntaxException{
// 1 获取文件配置信息
Configuration configuration = new Configuration();
// 2 判断是文件还是文件夹
FileStatus[] listStatus = fs.listStatus(new Path("/"));
for (FileStatus fileStatus : listStatus) {
// 如果是文件
if (fileStatus.isFile()) {
System.out.println("f:"+fileStatus.getPath().getName());
}else {
System.out.println("d:"+fileStatus.getPath().getName());
}
}
// 3 关闭资源
fs.close();
}
3. HDFS的I/O流操作
(1) HDFS文件上传
public static void put() throws Exception {
//1.获取配置信息
Configuration configuration = new Configuration();
//2.获取文件系统
//3.创建输入流
FileInputStream fis = new FileInputStream("/Users/luomingkui/Downloads/wc.input");
//4.创建输出流
FSDataOutputStream fos = fs.create(new Path("/luomk/wc.input"));
//5.流的对接
IOUtils.copyBytes(fis, fos, configuration);
//6.关闭资源
IOUtils.closeStream(fis);
IOUtils.closeStream(fos);
}
(2) HDFS文件下载
public static void get() throws URISyntaxException, IOException, InterruptedException {
//1.获取配置信息
Configuration configuration = new Configuration();
//2.获取文件系统
FileSystem fs = FileSystem.get(new URI("hdfs://hadoop103:9000"), configuration, "luomk");
//3.获取输入流
FSDataInputStream fis = fs.open(new Path("/luomk/wc1.input"));
//4.获取输出流
FileOutputStream fos = new FileOutputStream("/Users/luomingkui/Downloads/wc1.input");
//5.流的对接
IOUtils.copyBytes(fis, fos, configuration);
//6.关闭资源
IOUtils.closeStream(fis);
IOUtils.closeStream(fos);
}
(3) 定位文件读取
(1) 下载第一块
@Test
public void readFileSeek1() throws IOException, InterruptedException, URISyntaxException{
// 1 获取文件系统
Configuration configuration = new Configuration();
// 2 获取输入流
FSDataInputStream fis = fs.open(new Path("/hadoop-2.7.2.tar.gz"));
// 3 创建输出流
FileOutputStream fos = new FileOutputStream(new File("/Users/luomingkui/Downloads/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);
}
(2) 下载第二块
@Test
public void readFileSeek2() throws IOException, InterruptedException, URISyntaxException{
// 1 获取文件系统
Configuration configuration = new Configuration();
// 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("/Users/luomingkui/Downloads/hadoop-2.7.2.tar.gz.part2"));
// 5 流的对拷
IOUtils.copyBytes(fis, fos, configuration);
// 6 关闭资源
IOUtils.closeStream(fis);
IOUtils.closeStream(fos);
}
(3) 合并文件
在window命令窗口中执行
type hadoop-2.7.2.tar.gz.part2 >> hadoop-2.7.2.tar.gz.part1