blog/BigData-Notes
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

modify

Browse Source
This commit is contained in:
luoxiang
2019-05-01 12:46:02 +08:00
parent 5175fcce74
commit 55be9cfe4e
2 changed files with 372 additions and 369 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@@ -54,7 +54,7 @@
3. [集群资源管理器——YARN](https://github.com/heibaiying/BigData-Notes/blob/master/notes/Hadoop-YARN.md) 3. [集群资源管理器——YARN](https://github.com/heibaiying/BigData-Notes/blob/master/notes/Hadoop-YARN.md)
4. [Hadoop单机伪集群环境搭建](https://github.com/heibaiying/BigData-Notes/blob/master/notes/installation/hadoop%E5%8D%95%E6%9C%BA%E7%89%88%E6%9C%AC%E7%8E%AF%E5%A2%83%E6%90%AD%E5%BB%BA.md) 4. [Hadoop单机伪集群环境搭建](https://github.com/heibaiying/BigData-Notes/blob/master/notes/installation/hadoop%E5%8D%95%E6%9C%BA%E7%89%88%E6%9C%AC%E7%8E%AF%E5%A2%83%E6%90%AD%E5%BB%BA.md)
5. [HDFS常用Shell命令](https://github.com/heibaiying/BigData-Notes/blob/master/notes/HDFS常用Shell命令.md) 5. [HDFS常用Shell命令](https://github.com/heibaiying/BigData-Notes/blob/master/notes/HDFS常用Shell命令.md)
6. HDFS Java API详解 6. [HDFS Java API的使用](https://github.com/heibaiying/BigData-Notes/blob/master/notes/HDFS-Java-API.md)
## 二、Hive ## 二、Hive

739
notes/HDFS-Java-API.md
View File

@@ -1,5 +1,6 @@
# HDFS Java API # HDFS Java API
<nav>
<a href="#一-简介">一、 简介</a><br/> <a href="#一-简介">一、 简介</a><br/>
<a href="#二API的使用">二、API的使用</a><br/> <a href="#二API的使用">二、API的使用</a><br/>
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<a href="#21-FileSystem">2.1 FileSystem</a><br/> &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<a href="#21-FileSystem">2.1 FileSystem</a><br/>
@@ -18,370 +19,372 @@
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<a href="#214-查看文件的块信息">2.14 查看文件的块信息</a><br/> &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<a href="#214-查看文件的块信息">2.14 查看文件的块信息</a><br/>
</nav> </nav>
## 一、 简介
## 一、 简介
Hadoop提供了简单易用的Java API用于操作HDFS。通过这些API，我们可以通过编程来更灵活的操作HDFS。同时从编程体验上来说，这些API的设计确实非常人性化，基本上你只需要一行代码就能完成相应的操作。
Hadoop提供了简单易用的Java API用于操作HDFS。通过这些API，我们可以通过编程来更灵活的操作HDFS。同时从编程体验上来说，这些API的设计确实非常人性化，基本上你只需要一行代码就能完成相应的操作。
想要使用HDFS API，你只需要导入`hadoop-client`这个依赖包即可。以下关于API的操作我均使用单元测试的方法进行演示，完整的POM文件如下：
想要使用HDFS API，你只需要导入`hadoop-client`这个依赖包即可。以下关于API的操作我均使用单元测试的方法进行演示，完整的POM文件如下：
```xml
```xml <?xml version="1.0" encoding="UTF-8"?>
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0"
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion>
<modelVersion>4.0.0</modelVersion>
<groupId>com.heibaiying</groupId>
<groupId>com.heibaiying</groupId> <artifactId>hdfs-java-api</artifactId>
<artifactId>hdfs-java-api</artifactId> <version>1.0</version>
<version>1.0</version>
<properties>
<properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <hadoop.version>2.6.0-cdh5.15.2</hadoop.version>
<hadoop.version>2.6.0-cdh5.15.2</hadoop.version> </properties>
</properties>
<!---配置CDH仓库地址-->
<!---配置CDH仓库地址--> <repositories>
<repositories> <repository>
<repository> <id>cloudera</id>
<id>cloudera</id> <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
<url>https://repository.cloudera.com/artifactory/cloudera-repos/</url> </repository>
</repository> </repositories>
</repositories>
<dependencies>
<dependencies> <!--Hadoop-client-->
<!--Hadoop-client--> <dependency>
<dependency> <groupId>org.apache.hadoop</groupId>
<groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId>
<artifactId>hadoop-client</artifactId> <version>${hadoop.version}</version>
<version>${hadoop.version}</version> </dependency>
</dependency> <dependency>
<dependency> <groupId>junit</groupId>
<groupId>junit</groupId> <artifactId>junit</artifactId>
<artifactId>junit</artifactId> <version>4.12</version>
<version>4.12</version> <scope>test</scope>
<scope>test</scope> </dependency>
</dependency> </dependencies>
</dependencies>
</project>
</project> ```
```
## 二、API的使用
## 二、API的使用
### 2.1 FileSystem
### 2.1 FileSystem
FileSystem是操作HDFS的入口，通过FileSystem我们可以完成对HDFS上文件和目录的所有操作。在使用前需要获取它，这里由于之后的每个单元测试都需要用到FileSystem，所以使用`@Before`注解进行标注。
FileSystem是操作HDFS的入口，通过FileSystem我们可以完成对HDFS上文件和目录的所有操作。在使用前需要获取它，这里由于之后的每个单元测试都需要用到FileSystem，所以使用`@Before`注解进行标注。
```java
```java private static final String HDFS_PATH = "hdfs://192.168.0.106:8020";
private static final String HDFS_PATH = "hdfs://192.168.0.106:8020"; private static final String HDFS_USER = "root";
private static final String HDFS_USER = "root"; private static FileSystem fileSystem;
private static FileSystem fileSystem;
@Before
@Before public void prepare() {
public void prepare() { try {
try { Configuration configuration = new Configuration();
Configuration configuration = new Configuration(); // 这里我启动的是单节点的Hadoop,副本系数可以设置为1,不设置的话默认值为3
// 这里我启动的是单节点的Hadoop,副本系数可以设置为1,不设置的话默认值为3 configuration.set("dfs.replication", "1");
configuration.set("dfs.replication", "1"); fileSystem = FileSystem.get(new URI(HDFS_PATH), configuration, HDFS_USER);
fileSystem = FileSystem.get(new URI(HDFS_PATH), configuration, HDFS_USER); } catch (IOException e) {
} catch (IOException e) { e.printStackTrace();
e.printStackTrace(); } catch (InterruptedException e) {
} catch (InterruptedException e) { e.printStackTrace();
e.printStackTrace(); } catch (URISyntaxException e) {
} catch (URISyntaxException e) { e.printStackTrace();
e.printStackTrace(); }
} }
}
@After
@After public void destroy() {
public void destroy() { fileSystem = null;
fileSystem = null; }
} ```
```
### 2.2 创建目录
### 2.2 创建目录
这里目录可以是多级，支持递归创建。
这里目录可以是多级，支持递归创建。
```java
```java @Test
@Test public void mkDir() throws Exception {
public void mkDir() throws Exception { fileSystem.mkdirs(new Path("/hdfs-api/test0/"));
fileSystem.mkdirs(new Path("/hdfs-api/test0/")); }
} ```
```
### 2.3 创建指定权限的目录
### 2.3 创建指定权限的目录
这里`FsPermission(FsAction u, FsAction g, FsAction o)` 的三个参数分别对应创建者权限，同组其他用户权限，其他用户权限，可以使用`FsAction`枚举类中的值进行指定。
这里`FsPermission(FsAction u, FsAction g, FsAction o)` 的三个参数分别对应创建者权限，同组其他用户权限，其他用户权限，可以使用`FsAction`枚举类中的值进行指定。
```java
```java @Test
@Test public void mkDirWithPermission() throws Exception {
public void mkDirWithPermission() throws Exception { fileSystem.mkdirs(new Path("/hdfs-api/test1/"),
fileSystem.mkdirs(new Path("/hdfs-api/test1/"), new FsPermission(FsAction.READ_WRITE, FsAction.READ, FsAction.READ));
new FsPermission(FsAction.READ_WRITE, FsAction.READ, FsAction.READ)); }
} ```
```
### 2.4 创建文件，并写入内容
### 2.4 创建文件，并写入内容
```java
```java @Test
@Test public void create() throws Exception {
public void create() throws Exception { // 如果文件存在，默认会覆盖, 可以通过第二个参数进行控制。第三个参数可以控制使用缓冲区的大小
// 如果文件存在，默认会覆盖, 可以通过第二个参数进行控制。第三个参数可以控制使用缓冲区的大小 FSDataOutputStream out = fileSystem.create(new Path("/hdfs-api/test/a.txt"),
FSDataOutputStream out = fileSystem.create(new Path("/hdfs-api/test/a.txt"), true, 4096);
true, 4096); out.write("hello hadoop!".getBytes());
out.write("hello hadoop!".getBytes()); out.write("hello spark!".getBytes());
out.write("hello spark!".getBytes()); out.write("hello flink!".getBytes());
out.write("hello flink!".getBytes()); // 强制将缓冲区中内容刷出
// 强制将缓冲区中内容刷出 out.flush();
out.flush(); out.close();
out.close(); }
} ```
```
### 2.5 判断文件是否存在
### 2.5 判断文件是否存在
```java
```java @Test
@Test public void exist() throws Exception {
public void exist() throws Exception { boolean exists = fileSystem.exists(new Path("/hdfs-api/test/a.txt"));
boolean exists = fileSystem.exists(new Path("/hdfs-api/test/a.txt")); System.out.println(exists);
System.out.println(exists); }
} ```
```
### 2.6 查看文件内容
### 2.6 查看文件内容
这里我们查看的是一个小文件的内容，所以直接转换成字符串后输出，对于大文件，还是应该从输出流中读取数据，分批处理。
这里我们查看的是一个小文件的内容，所以直接转换成字符串后输出，对于大文件，还是应该从输出流中读取数据，分批处理。
```java
```java @Test
@Test public void readToString() throws Exception {
public void readToString() throws Exception { FSDataInputStream inputStream = fileSystem.open(new Path("/hdfs-api/test/a.txt"));
FSDataInputStream inputStream = fileSystem.open(new Path("/hdfs-api/test/a.txt")); String context = inputStreamToString(inputStream, "utf-8");
String context = inputStreamToString(inputStream, "utf-8"); System.out.println(context);
System.out.println(context); }
} ```
```
inputStreamToString 是一个自定义方法，实现如下：
inputStreamToString 是一个自定义方法，实现如下：
```java
```java /**
/** * 把输入流转换为指定编码的字符
* 把输入流转换为指定编码的字符 *
* * @param inputStream 输入流
* @param inputStream 输入流 * @param encode 指定编码类型
* @param encode 指定编码类型 */
*/ private static String inputStreamToString(InputStream inputStream, String encode) {
private static String inputStreamToString(InputStream inputStream, String encode) { try {
try { if (encode == null || ("".equals(encode))) {
if (encode == null || ("".equals(encode))) { encode = "utf-8";
encode = "utf-8"; }
} BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, encode));
BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, encode)); StringBuilder builder = new StringBuilder();
StringBuilder builder = new StringBuilder(); String str = "";
String str = ""; while ((str = reader.readLine()) != null) {
while ((str = reader.readLine()) != null) { builder.append(str).append("\n");
builder.append(str).append("\n"); }
} return builder.toString();
return builder.toString(); } catch (IOException e) {
} catch (IOException e) { e.printStackTrace();
e.printStackTrace(); }
} return null;
return null; }
} ```
```
### 2.7 文件重命名
### 2.7 文件重命名
```java
```java @Test
@Test public void rename() throws Exception {
public void rename() throws Exception { Path oldPath = new Path("/hdfs-api/test/a.txt");
Path oldPath = new Path("/hdfs-api/test/a.txt"); Path newPath = new Path("/hdfs-api/test/b.txt");
Path newPath = new Path("/hdfs-api/test/b.txt"); boolean result = fileSystem.rename(oldPath, newPath);
boolean result = fileSystem.rename(oldPath, newPath); System.out.println(result);
System.out.println(result); }
} ```
```
### 2.8 删除目录或文件
### 2.8 删除目录或文件
```java
```java public void delete() throws Exception {
public void delete() throws Exception { /*
/* * 第二个参数代表是否递归删除
* 第二个参数代表是否递归删除 * + 如果path是一个目录且递归删除为true, 则删除该目录及其中所有文件;
* + 如果path是一个目录且递归删除为true, 则删除该目录及其中所有文件; * + 如果path是一个目录但递归删除为false,则会则抛出异常。
* + 如果path是一个目录但递归删除为false,则会则抛出异常。 */
*/ boolean result = fileSystem.delete(new Path("/hdfs-api/test/b.txt"), true);
boolean result = fileSystem.delete(new Path("/hdfs-api/test/b.txt"), true); System.out.println(result);
System.out.println(result); }
} ```
```
### 2.9 上传文件到HDFS
### 2.9 上传文件到HDFS
```java
```java @Test
@Test public void copyFromLocalFile() throws Exception {
public void copyFromLocalFile() throws Exception { // 如果指定的是目录，则会把目录及其中的文件都复制到指定目录下
// 如果指定的是目录，则会把目录及其中的文件都复制到指定目录下 Path src = new Path("D:\\BigData-Notes\\notes\\installation");
Path src = new Path("D:\\BigData-Notes\\notes\\installation"); Path dst = new Path("/hdfs-api/test/");
Path dst = new Path("/hdfs-api/test/"); fileSystem.copyFromLocalFile(src, dst);
fileSystem.copyFromLocalFile(src, dst); }
} ```
```
### 2.10 上传大文件并显示上传进度
### 2.10 上传大文件并显示上传进度
```java
```java @Test
@Test public void copyFromLocalBigFile() throws Exception {
public void copyFromLocalBigFile() throws Exception {
File file = new File("D:\\kafka.tgz");
File file = new File("D:\\kafka.tgz"); final float fileSize = file.length();
final float fileSize = file.length(); InputStream in = new BufferedInputStream(new FileInputStream(file));
InputStream in = new BufferedInputStream(new FileInputStream(file));
FSDataOutputStream out = fileSystem.create(new Path("/hdfs-api/test/kafka5.tgz"),
FSDataOutputStream out = fileSystem.create(new Path("/hdfs-api/test/kafka5.tgz"), new Progressable() {
new Progressable() { long fileCount = 0;
long fileCount = 0;
public void progress() {
public void progress() { fileCount++;
fileCount++; // progress方法每上传大约64KB的数据后就会被调用一次
// progress方法每上传大约64KB的数据后就会被调用一次 System.out.println("上传进度：" + (fileCount * 64 * 1024 / fileSize) * 100 + " %");
System.out.println("上传进度：" + (fileCount * 64 * 1024 / fileSize) * 100 + " %"); }
} });
});
IOUtils.copyBytes(in, out, 4096);
IOUtils.copyBytes(in, out, 4096);
}
} ```
```
### 2.11 从HDFS上下载文件
### 2.11 从HDFS上下载文件
```java
```java @Test
@Test public void copyToLocalFile() throws Exception {
public void copyToLocalFile() throws Exception { Path src = new Path("/hdfs-api/test/kafka.tgz");
Path src = new Path("/hdfs-api/test/kafka.tgz"); Path dst = new Path("D:\\app\\");
Path dst = new Path("D:\\app\\"); /*
/* * 第一个参数控制下载完成后是否删除源文件,默认是true,即删除;
* 第一个参数控制下载完成后是否删除源文件,默认是true,即删除; * 最后一个参数表示是否将RawLocalFileSystem用作本地文件系统;
* 最后一个参数表示是否将RawLocalFileSystem用作本地文件系统; * RawLocalFileSystem默认为false,通常情况下可以不设置,
* RawLocalFileSystem默认为false,通常情况下可以不设置, * 但如果你在执行时候抛出NullPointerException异常,则代表你的文件系统与程序可能存在不兼容的情况(window下常见),
* 但如果你在执行时候抛出NullPointerException异常,则代表你的文件系统与程序可能存在不兼容的情况(window下常见), * 此时可以将RawLocalFileSystem设置为true
* 此时可以将RawLocalFileSystem设置为true */
*/ fileSystem.copyToLocalFile(false, src, dst, true);
fileSystem.copyToLocalFile(false, src, dst, true); }
} ```
```
### 2.12 查看指定目录下所有文件的信息
### 2.12 查看指定目录下所有文件的信息
```java
```java public void listFiles() throws Exception {
public void listFiles() throws Exception { FileStatus[] statuses = fileSystem.listStatus(new Path("/hdfs-api"));
FileStatus[] statuses = fileSystem.listStatus(new Path("/hdfs-api")); for (FileStatus fileStatus : statuses) {
for (FileStatus fileStatus : statuses) { //fileStatus的toString方法被重写过，直接打印可以看到所有信息
//fileStatus的toString方法被重写过，直接打印可以看到所有信息 System.out.println(fileStatus.toString());
System.out.println(fileStatus.toString()); }
} }
} ```
```
FileStatus中包含了文件的基本信息，比如文件路径，是否是文件夹，修改时间，访问时间，所有者，所属组，文件权限，是否是符号链接等，输出内容示例如下(这里为了直观，我对输出进行了换行显示)：
FileStatus中包含了文件的基本信息，比如文件路径，是否是文件夹，修改时间，访问时间，所有者，所属组，文件权限，是否是符号链接等，输出内容示例如下(这里为了直观，我对输出进行了换行显示)：
```properties
```properties FileStatus{
FileStatus{ path=hdfs://192.168.0.106:8020/hdfs-api/test;
path=hdfs://192.168.0.106:8020/hdfs-api/test; isDirectory=true;
isDirectory=true; modification_time=1556680796191;
modification_time=1556680796191; access_time=0;
access_time=0; owner=root;
owner=root; group=supergroup;
group=supergroup; permission=rwxr-xr-x;
permission=rwxr-xr-x; isSymlink=false
isSymlink=false }
} ```
```
### 2.13 递归查看指定目录下所有文件的信息
### 2.13 递归查看指定目录下所有文件的信息
```java
```java @Test
@Test public void listFilesRecursive() throws Exception {
public void listFilesRecursive() throws Exception { RemoteIterator<LocatedFileStatus> files = fileSystem.listFiles(new Path("/hbase"), true);
RemoteIterator<LocatedFileStatus> files = fileSystem.listFiles(new Path("/hbase"), true); while (files.hasNext()) {
while (files.hasNext()) { System.out.println(files.next());
System.out.println(files.next()); }
} }
} ```
```
这里输出和上面类似，只是多了文本大小，副本系数，块大小等信息。
这里输出和上面类似，只是多了文本大小，副本系数，块大小等信息。
```properties
```properties LocatedFileStatus{
LocatedFileStatus{ path=hdfs://192.168.0.106:8020/hbase/hbase.version;
path=hdfs://192.168.0.106:8020/hbase/hbase.version; isDirectory=false;
isDirectory=false; length=7;
length=7; replication=1;
replication=1; blocksize=134217728;
blocksize=134217728; modification_time=1554129052916;
modification_time=1554129052916; access_time=1554902661455;
access_time=1554902661455; owner=root; group=supergroup;
owner=root; group=supergroup; permission=rw-r--r--;
permission=rw-r--r--; isSymlink=false}
isSymlink=false} ```
```
### 2.14 查看文件的块信息
### 2.14 查看文件的块信息
```java
```java @Test
@Test public void getFileBlockLocations() throws Exception {
public void getFileBlockLocations() throws Exception {
FileStatus fileStatus = fileSystem.getFileStatus(new Path("/hdfs-api/test/kafka.tgz"));
FileStatus fileStatus = fileSystem.getFileStatus(new Path("/hdfs-api/test/kafka.tgz")); BlockLocation[] blocks = fileSystem.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
BlockLocation[] blocks = fileSystem.getFileBlockLocations(fileStatus, 0, fileStatus.getLen()); for (BlockLocation block : blocks) {
for (BlockLocation block : blocks) { System.out.println(block);
System.out.println(block); }
} }
} ```
```
块输出信息比较简单，第一个值是文件的起始偏移量(offset)，第二个值是文件大小(length)，第三个是块所在的主机名(hosts)。
块输出信息比较简单，第一个值是文件的起始偏移量(offset)，第二个值是文件大小(length)，第三个是块所在的主机名(hosts)。
```
``` 0,57028557,hadoop001
0,57028557,hadoop001 ```
```
这里我上传的文件比较小，只有57M(小于128M)，且程序中设置了副本系数为1，所有只有一个块信息，如果文件很大，则这里会输出文件所有块的信息。
这里我上传的文件比较小，只有57M(小于128M)，且程序中设置了副本系数为1，所有只有一个块信息，如果文件很大，则这里会输出文件所有块的信息。
<br/>
<br/>
**以上所有测试用例源码可以从本仓库进行下载**：[HDFS Java API](https://github.com/heibaiying/BigData-Notes/tree/master/code/Hadoop/hdfs-java-api) **以上所有测试用例源码可以从本仓库进行下载**：[HDFS Java API](https://github.com/heibaiying/BigData-Notes/tree/master/code/Hadoop/hdfs-java-api)