HDFS的Java API开发详解（文件的创建上传下载删除、IO流操作HDFS、小文件合并）
目录
一、准备工作
-
windows操作系统需要配置一下hadoop环境
-
mac本质上是unix系统，不需要配置
第一步：windows中的hadoop环境配置
-
windows操作系统需要配置一下hadoop环境
-
mac本质上是unix系统，不需要配置
1. 解压Hadoop安装包压缩文件到一个没有中文没有空格的目录下，类似下图路径
2. 然后在windows当中配置hadoop的环境变量
3. bin、sbin目录添加到path中
4. 然后将下图中的hadoop.dll文件拷贝到C:\Windows\System32
5. 将Linux上安装部署好的hadoop集群的以下5个配置文件core-site.xml、hdfs-site.xml、mapred-site.xml、yarn-site.xml、slaves，拷贝到windows下hadoop的C:\hadoop-2.6.0-cdh5.14.0\etc\hadoop目录下
6. cmd中运行hadoop，出现如下效果
在Linux虚拟机中hdfs集群启动的前提下，cmd中运行hdfs dfs -ls /，查询出hdfs根目录的内容
注意：如果没有配置好windows的hadoop的环境变量，在windows下用IDEA编程时，会报以下错误
第二步：创建maven工程并导入jar包
-
由于cdh版本的所有的软件涉及版权的问题，所以并没有将所有的jar包托管到maven仓库当中去，而是托管在了CDH自己的服务器上面，所以我们默认去maven的仓库下载不到，需要自己手动的添加repository去CDH仓库进行下载，以下两个地址是官方文档说明，请仔细查阅
https://www.cloudera.com/documentation/enterprise/release-notes/topics/cdh_vd_cdh5_maven_repo.html
<repositories>
<repository>
<id>cloudera</id>
<url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
</repository>
</repositories>
<dependencies>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>2.6.0-mr1-cdh5.14.2</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>2.6.0-cdh5.14.2</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-hdfs</artifactId>
<version>2.6.0-cdh5.14.2</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-mapreduce-client-core</artifactId>
<version>2.6.0-cdh5.14.2</version>
</dependency>
<!-- https://mvnrepository.com/artifact/junit/junit -->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.11</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testng</groupId>
<artifactId>testng</artifactId>
<version>RELEASE</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.0</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
<encoding>UTF-8</encoding>
<!-- <verbal>true</verbal>-->
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>2.4.3</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<minimizeJar>true</minimizeJar>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>二、Java API开发实操
package com.xsluo;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
public class HdfsJavaAPI {
public static void main(String[] args) throws IOException, URISyntaxException, InterruptedException {
HdfsJavaAPI o = new HdfsJavaAPI();
//1.创建
o.mkdirToHdfs();
//2.上传
o.uploadFile();
//3.下载
o.downloadFile();
//4.删除
o.deleteFile();
//5.重命名
o.renameFile();
//6.查看文件的描述信息
o.testlistFiles();
//7.IO流上传
o.ioPutFile();
//8.IO流下载
o.ioGetFile();
//9.合并小文件
o.mergeSmallFiles();
}
/**
* 获取HDFS文件系统
* @return
* @throws IOException
*/
public FileSystem getFileSystem() throws IOException {
Configuration configuration = new Configuration();
configuration.set("fs.defaultFS","hdfs://node01:8020");
FileSystem fileSystem = FileSystem.get(configuration);
return fileSystem;
}
/**
* 创建文件夹
* @throws IOException
*/
public void mkdirToHdfs() throws IOException {
FileSystem fileSystem = getFileSystem();
boolean b = fileSystem.mkdirs(new Path("/xsluo/dir1"));//若目录已经存在，则创建失败，返回false
if (b){
System.out.println("/xsluo/dir1创建成功！");
}else {
System.out.println("/xsluo/dir1可能已存在，创建失败！");
}
fileSystem.close();
}
/**
* 文件上传
* @throws IOException
*/
public void uploadFile() throws IOException {
FileSystem fileSystem = getFileSystem();
fileSystem.copyFromLocalFile(
new Path("file:///F:\\testDatas\\a.txt"),
new Path("hdfs://node01:8020/xsluo"));//hdfs路径也可以直接写成/xsluo/dir1
fileSystem.close();
}
/**
* 文件下载
* @throws IOException
*/
public void downloadFile() throws IOException {
FileSystem fileSystem = getFileSystem();
fileSystem.copyToLocalFile(new Path("hdfs://node01:8020/xsluo/a.txt"),new Path("file:///F:\\testDatas\\a2.txt"));
fileSystem.close();
}
/**
* 文件删除
* @throws IOException
*/
public void deleteFile() throws IOException {
FileSystem fileSystem = getFileSystem();
fileSystem.delete(new Path("hdfs://node01:8020/xsluo/b.txt"),true);
fileSystem.close();
}
/**
* 文件重命名
* @throws IOException
*/
public void renameFile() throws IOException {
FileSystem fileSystem = getFileSystem();
fileSystem.rename(new Path("/xsluo/a.txt"),new Path("/xsluo/a_new.txt"));
fileSystem.close();
}
/**
* 文件相关信息查看
* @throws IOException
*/
public void testlistFiles() throws IOException {
FileSystem fileSystem = getFileSystem();
//获取文件详情
RemoteIterator<LocatedFileStatus> listFiles = fileSystem.listFiles(new Path("/xsluo"), true);
while (listFiles.hasNext()){
LocatedFileStatus status = listFiles.next();
//输出详情
System.out.println(status.getPath().getName());//文件名称
System.out.println(status.getLen());//长度
System.out.println(status.getPermission());//权限
System.out.println(status.getOwner());//所属用户
System.out.println(status.getGroup());//分组
System.out.println(status.getModificationTime());//修改时间
//获取存储的块信息
BlockLocation[] blockLocations = status.getBlockLocations();
for (BlockLocation blockLocation : blockLocations) {
//获取块存储的主机节点
String[] hosts = blockLocation.getHosts();
for (String host : hosts) {
System.out.println(host);
}
}
}
fileSystem.close();
}
/**
* IO流操作hdfs文件：上传
* @throws IOException
*/
public void ioPutFile() throws IOException {
// 1 获取文件系统
FileSystem fileSystem = getFileSystem();
// 2 创建输入流；路径前不需要加file:///，否则报错
FileInputStream fis = new FileInputStream(new File("F:\\testDatas\\hhh.txt"));
// 3 创建输出流
FSDataOutputStream fos = fileSystem.create(new Path("hdfs://node01:8020/xsluo/hhh.txt"));
// 4 流对拷 org.apache.commons.io.IOUtils
IOUtils.copy(fis,fos);
// 5 关闭资源
IOUtils.closeQuietly(fos);
IOUtils.closeQuietly(fis);
fileSystem.close();
}
/**
* IO流操作hdfs文件：下载
* @throws IOException
*/
public void ioGetFile() throws IOException {
// 1 获取文件系统
FileSystem fileSystem = getFileSystem();
// 2 创建输入流
FSDataInputStream fis = fileSystem.open(new Path("/xsluo/a_new.txt"));
// 3 创建输出流
FileOutputStream fos = new FileOutputStream(new File("F:\\testDatas\\a_new.txt"));
// 4 流对拷
IOUtils.copy(fis,fos);
// 5 关闭资源
IOUtils.closeQuietly(fos);
IOUtils.closeQuietly(fis);
fileSystem.close();
}
/**
* 小文件合并
* @throws URISyntaxException
* @throws IOException
* @throws InterruptedException
*/
public void mergeSmallFiles() throws URISyntaxException, IOException, InterruptedException {
//获取分布式文件系统hdfs；第三个参数指定hdfs的用户
FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration(), "hadoop");
FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path("hdfs://node01:8020/xsluo/bigfile.txt"));
//读取所有本地小文件，写入到hdfs的大文件里面去
//获取本地文件系统 localFileSystem
LocalFileSystem localFileSystem = FileSystem.getLocal(new Configuration());
//读取本地的小文件们
FileStatus[] fileStatuses = localFileSystem.listStatus(new Path("F:\\testDatas"));
for (FileStatus fileStatus : fileStatuses) {
//获取每一个本地小文件的路径
Path path = fileStatus.getPath();
//读取本地小文件
FSDataInputStream fsDataInputStream = localFileSystem.open(path);
IOUtils.copy(fsDataInputStream,fsDataOutputStream);
IOUtils.closeQuietly(fsDataInputStream);
}
IOUtils.closeQuietly(fsDataOutputStream);
localFileSystem.close();
fileSystem.close();
}
}