HDFS之Java客户端操作_连接hdfs的客户端

HDFS之Java客户端操作

---

---

写在前面

• Hadoop版本：Hadoop-3.1.3
• Linux版本：CentOS7.5
• IDE工具：IntelliJ IDEA（Windows环境下）

HDFS的Java客户端操作也是入门Hadoop开发的学习重点

准备Windows关于Hadoop的开发环境

下载依赖

https://github.com/steveloughran/winutils

根据自己的Hadoop安装版本选择对应版本的依赖，如果需要更高版本的依赖，请从以下地址下载

https://github.com/cdarlint/winutils/tree/master/hadoop-3.2.0/bin

将下载的文件拷贝到你存储Hadoop依赖的路径中（比如是E:\Development目录下）

配置HADOOP_HOME环境变量

选择系统变量，跟JAVA_HOME的配置是一样的

配置Path环境变量

• 编辑环境变量

需要重启电脑，使得环境生效

创建Maven工程

XML文件

创建一个Maven工程 HdfsClientDemo，并导入相应的依赖坐标+日志添加，pom.xml文件内容如下：

<dependencies>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
    </dependency>
    <dependency>
        <groupId>org.apache.logging.log4j</groupId>
        <artifactId>log4j-slf4j-impl</artifactId>
        <version>2.12.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>3.1.3</version>
    </dependency>
</dependencies>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

在项目的 src/main/resources 目录下，新建一个文件，命名为 log4j2.xml ，在文件中填入以下内容：

<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="error" strict="true" name="XMLConfig">
    <Appenders>
        <!-- 类型名为Console，名称为必须属性 -->
        <Appender type="Console" name="STDOUT">
            <!-- 布局为PatternLayout的方式，
            输出样式为[INFO] [2018-01-22 17:34:01][org.test.Console]I'm here -->
            <Layout type="PatternLayout"
                    pattern="[%p] [%d{yyyy-MM-dd HH:mm:ss}][%c{10}]%m%n" />
        </Appender>

    </Appenders>

    <Loggers>
        <!-- 可加性为false -->
        <Logger name="test" level="info" additivity="false">
            <AppenderRef ref="STDOUT" />
        </Logger>

        <!-- root loggerConfig设置 -->
        <Root level="info">
            <AppenderRef ref="STDOUT" />
        </Root>
    </Loggers>
</Configuration>

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

创建新的Package

包名：cn.whybigdata.hdfs

创建HdfsClient类

尝试连接Hadoop集群并创建新的目录/xiyouji/sunwukong

public class HdfsClient{	
    @Test
    public void testMkdirs() throws IOException, InterruptedException, URISyntaxException{		
		// 1 获取文件系统
		Configuration configuration = new Configuration();
		// 配置在集群上运行
		// configuration.set("fs.defaultFS", "hdfs://node01:9820");
		// FileSystem fs = FileSystem.get(configuration);
		FileSystem fs = FileSystem.get(new URI("hdfs://node01:9820"), configuration, "whybigdata");	
		// 2 创建目录
		fs.mkdirs(new Path("/xiyouji/sunwukong"));		
		// 3 关闭资源
		fs.close();
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

执行程序

运行时需要配置用户名称whybigdata

客户端去操作HDFS时，是有一个用户身份的。默认情况下，HDFS客户端API会从JVM中获取一个参数来作为自己的用户身份：-DHADOOP_USER_NAME=whybigdata，whybigdata为用户名称。

HDFS的API操作

常见的API操作包括：文件上传、下载；删除文件和目录；文件重命名、移动、查看文件详情；HDFS文件和文件夹的判别。带如下所示：

package cn.whybigdata.hdfs;

/**
 * @author whybigdata
 * @create 2022-12-17-16:21
 */

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.net.URI;


/**
 * 1. 和 HDFS 建立连接
 * 2. 调用 API 完成具体功能
 * 3. 关闭连接
 */
public class HdfsClientTest {

    private FileSystem fs;

    /**
     *  TODO 判断是文件还是目录
     * @throws IOException
     */
    @Test
    public void testFileStatus() throws IOException{
        FileStatus[] fileStatuses = fs.listStatus(new Path("/"));
        for (FileStatus status : fileStatuses) {
            if (status.isDirectory()) {
                System.out.println("DIR:" + status.getPath().getName());
            } else {
                System.out.println("FILE:" + status.getPath().getName());
            }
        }
    }


    /**
     *  TODO 查看文件详情
     * @throws IOException
     */
    @Test
    public void testListFiles() throws IOException{
        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);
        while (listFiles.hasNext()) {
            LocatedFileStatus fileStatus = listFiles.next();
            System.out.println("文件名" + fileStatus.getPath().getName());
            System.out.println("块大小" + fileStatus.getBlockSize());
            System.out.println("副本数" + fileStatus.getReplication());
            System.out.println("权限信息：" + fileStatus.getPermission());
        }
    }


    /**
     *  TODO HDFS文件更名和移动
     * @throws IOException
     */
    @Test
    public void testRename() throws IOException{
        // 移动文件
//        fs.rename(new Path("/sanguo/liubie.txt"), new Path("/client_test"));
        // 更名
        fs.rename(new Path("/client_test/liubie.txt"), new Path("/client_test/xiaoqiao.txt"));
    }


    /**
     * TODO 删除文件和目录
     *
     */
    @Test
    public void testDelete() throws IOException {
        fs.delete(new Path("/client_test/hello.txt"), true);
    }


    /**
     * TODO 下载文件
     *
     */
    @Test
    public void testCopyToLocal() throws IOException {
        fs.copyToLocalFile(false, new Path("/client_test/hello.txt"),
                new Path("E:\\Test"), true);
    }

    /**
     * TODO 上传文件
     *  测试配置的优先级 Configuration > hdfs-site.xml > hdfs-default.xml
     */
    @Test
    public void testCopyFromLocal() throws IOException {
        fs.copyFromLocalFile(false, true,
                new Path("E:\\hello.txt"),
                new Path("/client_test"));
    }



    /**
     * TODO 获取 HDFS 对象
     * @throws IOException
     * @throws InterruptedException
     */
    @Before
    public void init() throws IOException, InterruptedException {
        // HDFS访问路径 hdfs://node01:9820
        URI uri = URI.create("hdfs://node01:9820");
        // conf : 配置对象
        Configuration conf = new Configuration();
        conf.set("dfs.replication", "6");
        // user : 操作用户（用哪个用户操作HDFS）
        String user = "whybigdata";
        // 获取HDFS的客户端连接（文件系统对象）
        fs = FileSystem.get(uri, conf, user);
    }


    /**
     * TODO 关闭资源
     */
    @After
    public void close() throws IOException {
       fs.close();
    }

    /**
     *  TODO 获取HDFS的客户端连接
     *  uri : HDFS访问路径 hdfs://node01:9820
     *  conf : 配置对象
     *  user : 操作用户（用哪个用户操作HDFS）
     */
    @Test
    public void testCreateHdfsClient() throws IOException, InterruptedException {
        // HDFS访问路径 hdfs://node01:9820
        URI uri = URI.create("hdfs://node01:9820");
        // conf : 配置对象
        Configuration conf = new Configuration();
        // user : 操作用户（用哪个用户操作HDFS）
        String user = "whybigdata";
        // 获取HDFS的客户端连接（文件系统对象）
        FileSystem fileSystem = FileSystem.get(uri, conf, user);
        System.out.println(fileSystem.getClass().getName());
        // 关闭资源
        fileSystem.close();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158

注意：

• 获取HDFS对象操作封装为一个 init() 方法
• HDFS文件更名和移动：调用的是同一个方法rename()
• 使用之后记得要关闭资源fileSystem.close()

全文结束！