大数据学习06：Java访问HDFS_java普通类加载hdfs文件

一、HDFS常用Java API

1、org.apache.hadoop.fs.FileSystem
一个通用文件系统的抽象基类，可被分布式文件系统继承。

2、org.apache.hadoop.fs.FileStatus
文件状态接口，用于向客户端展示系统中文件和目录的元数据。具体包括文件大小、块大小、副本信息、所有者、修改时间等，可通过FileSystem.listStatus()方法获得具体的实例对象。

3、org.apache.hadoop.fs.FileDataInputStream
文件输入流，用于读取Hadoop文件。

4、org.apache.hadoop.fs.FileDataOutputStream
文件输出流，用于写Hadoop文件。

5、org.apache.hadoop.fs.Configuration
访问配置项，所有配置项的值，如果在core-site.xml中有对应的配置，则以core-site.xml为准。

6、org.apache.hadoop.fs.Path
路径，用于表示Hadoop文件系统中的一个文件或一个目录的路径。

7、org.apache.hadoop.fs.PathFilter
路径过滤器接口，通过实现方法PathFilter.accept(Path path)来判断是否接收路径path表示的文件或目录。

8. HDFS中的主要涉及到的类

8.1 Configuration类

该类的对象封装了客户端或者服务器的配置

8.2 FileSystem类

该类的对象是一个文件系统对象，可以用该对象的一些方法对文件进行操作。其get()方法有两种：一个是只含有一个Configuration参数；另一个含有两个参数，分别是：URI和Configuration。

8.3 FSDataInputStream和FSDataOutputStream

这两个类是HDFS中的输入/输出流，分别通过FileSystem的open方法和create方法获得。

二、编写Java程序访问HDFS

1）启动虚拟机的HDFS服务
说明：虚拟机的地址是192.168.10.129主机名是hadoop129

三、HDFS上操作文件

上传文件

package net.qing.hdfs;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.net.URI;

public class createFolder {
    public static void main(String[] args)throws Exception {
        //创建配置对象
        Configuration conf = new Configuration();
        //定义uri字符串
        String uri ="hdfs://hadoop129:9000";
        //创建文件系统对象
        FileSystem fs = FileSystem.get(new URI(uri),conf);
        //创建路径对象
        Path path = new Path(uri + "/park/success.txt");
        //创建文件
        boolean result = fs.createNewFile(path);
        //判断是否创建成功
        if(result) {
            System.out.println("文件{" + path + "}创建成功！");
        }else {
            System.out.println("文件{" + path + "}创建失败！");
        }

    }
}
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

利用HDFS Explorer软件来查看：

再次运行，发现报错，因为success.txt已经存在

利用fsck命令查看这个文件：

利用web来查看

四、写入HDFS文件

1、创建WriteFileOnHDFS类：

package net.qing.hdfs;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.junit.Test;
import java.net.URI;
public class WriteFileOnHDFS {
    @Test
    public void write1() throws Exception {
        // 创建配置对象
        Configuration conf = new Configuration();
        // 定义uri字符串
        String uri = "hdfs://hadoop129:9000";
        // 创建文件系统对象
        FileSystem fs = FileSystem.get(new URI(uri), conf);
        // 创建路径对象（文件或目录）
        Path path = new Path(uri + "/park/success.txt");
        // 创建文件输出流
        FSDataOutputStream out = fs.create(path);
        // 写数据
        out.write("Hello Hadoop！！卿加波 write。。。".getBytes());
        // 提示用户写文件成功
        System.out.println("文件[" + path + "]写入成功！");
        // 刷新输出流
        out.flush();
        // 关闭输出流
        out.flush();
        // 关闭文件
        fs.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

利用HDFS Explorer查看success.txt：

将/park目录下的文件全部删除：


将本地文件写入hdfs文件

@Test
    public void write2() throws Exception {
        Configuration conf = new Configuration();
        String uri = "hdfs://hadoop129:9000";
        FileSystem fs = FileSystem.get(new URI(uri), conf);
        Path path = new Path(uri + "/park/text.txt");
        FSDataOutputStream out = fs.create(path);
        FileInputStream in = new FileInputStream("text.txt");
        BufferedReader br = new BufferedReader(new InputStreamReader(in));
        String nextLine = "";
        while ((nextLine = br.readLine()) != null) {
            System.out.println(nextLine);
            out.write(nextLine.getBytes());
        }
        System.out.println("文件[text.txt]写入文件[" + path + "]！");
        in.close();
        out.close();
        fs.close();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

HDFS Explorer查看：

五、读取hdfs文件

创建ReadFileOnHDFS类

（1）读取HDFS文件直接在控制台显示
准备读取hdfs://hadoop129:9000/park/test.txt文件：

@Test
    public void read1() throws Exception {
        //创建配置对象
        Configuration conf = new Configuration();
        //定义urizfc
        String uri = "hdfs://hadoop129:9000";
        //创建文件系统对象
        FileSystem fs = FileSystem.get(new URI(uri), conf);
        //创建路径对象（文件或目录）
        Path path = new Path(uri + "/park/text.txt");
        //创建文件输入流
        FSDataInputStream in = fs .open(path);
        //读取文件在控制台显示
        BufferedReader br = new BufferedReader(new InputStreamReader(in));
        String nexLine ="";
        while ((nexLine = br.readLine() )!= null) {
            System.out.println(nexLine);
        }
        //关闭文件系统
        fs.close();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

代码简化后

（2）读取HDFS文件，保存为本地文件
任务：将/park/text.txt下载到项目下/download目录里。
首先在项目目录创建download文件夹

@Test
    public void read2() throws Exception {
        Configuration conf = new Configuration();
        String uri = "hdfs://hadoop129:9000";
        FileSystem fs = FileSystem.get(new URI(uri), conf);
        Path path = new Path(uri + "/park/text.txt");
        FSDataInputStream in = fs.open(path);
        FileOutputStream out = new FileOutputStream("download/text.txt");
        IOUtils.copyBytes(in, out, conf);
        System.out.println("文件[" + path + "]下载到本地文件[download/text.txt]！" );
        fs.close();
    }
1
2
3
4
5
6
7
8
9
10
11
12

六、重命名目录或文件

创建RenameDirOrFile类：
（1）重命名目录

将park更名为park01

@Test
    public void renameDir() throws Exception {
        Configuration conf = new Configuration();
        String uri = "hdfs://hadoop129:9000";
        FileSystem fs = FileSystem.get(new URI(uri), conf);
        Path path1 = new Path("/park");
        Path path2 = new Path("/park01");
        fs.rename(path1, path2);
        System.out.println("目录[" + path1.getName() + "]重命名为目录[" + path2.getName() + "]！");
    }
1
2
3
4
5
6
7
8
9
10

（2）重命名文件

@Test
    public void renamefile()throws Exception {
        Configuration conf = new Configuration();
        String uri = "hdfs://hadoop129:9000";
        FileSystem fs = FileSystem.get(new URI(uri), conf);
        Path path1 = new Path("/park01/text.txt");
        Path path2 = new Path("/park01/text007.txt");
        fs.rename(path1, path2);
        System.out.println("文件[" + path1 + "]重命名为文件[" + path2 + "]！");
    }
1
2
3
4
5
6
7
8
9
10

七、显示文件列表

先上传一些文件到park01

创建ListHDFSFiles类：

1）显示根目录下所有文件的全部信息

@Test
    public void list1() throws Exception {
        Configuration conf = new Configuration();
        String uri = "hdfs://hadoop129:9000";
        FileSystem fs = FileSystem.get(new URI(uri), conf);
        RemoteIterator<LocatedFileStatus> ri = fs.listFiles(new Path("/"), true);
        while (ri.hasNext()) {
            System.out.println(ri.next());
        }
    }
1
2
3
4
5
6
7
8
9
10

运行程序，结果如下：

上述文件状态对象封装的有关信息，可以通过相应的方法来获取，比如getPath()方法就可以获取路径信息。
2）只显示文件的路径信息

@Test
public void list2() throws Exception {
    Configuration conf = new Configuration();
    String uri = "hdfs://hadoop129:9000";
    FileSystem fs = FileSystem.get(new URI(uri), conf);
    RemoteIterator<LocatedFileStatus> ri = fs.listFiles(new Path("/"), true);
    while (ri.hasNext()) {
        System.out.println(ri.next().getPath());
    }
}
1
2
3
4
5
6
7
8
9
10

八、获取文件块信息

上传一个大于128MB的文件，比如hadoop-2.10.0.tar.gz，到/park01目录：
创建GetBlockLocations类：

package net.qing.hdfs;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.net.URI;

public class GetBlockLocations {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String uri = "hdfs://hadoop129:9000";
        FileSystem fs = FileSystem.get(new URI(uri), conf);
        Path path = new Path("/park01/hadoop-2.5.0-cdh5.3.6.tar.gz");
        BlockLocation[] blks = fs.getFileBlockLocations(path, 0, Integer.MAX_VALUE);
        for (BlockLocation blk : blks) {
            System.out.println(blk);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

由此可见，hadoop-2.10.0.tar.gz被hadoop物理切分成两块，第一块长度为134217728字节（128MB），第二块长度为134217728字节（128MB），第三块长度为123680277字节（118MB）。

利用Web界面也可以查看文件分块信息：

九、创建目录文件

HDFS Explorer查看原目录

创建MakeDirOnHDFS类

package net.qing.hdfs;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import java.net.URI;
public class MakeDirOnHDFS {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String uri = "hdfs://hadoop129:9000";
        FileSystem fs = FileSystem.get(new URI(uri), conf);
        Path path = new Path("/park02");
        boolean result = fs.mkdirs(path);
        if (result) {
            System.out.println("目录[" + path +"]创建成功！");
        } else {
            System.out.println("目录[" + path +"]创建失败！");
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

任务：创建/park02目录。

利用HDFS Explorer查看

再运行程序，/park01已经存在，但是仍然提示用户创建成功：

发现文件系统出问题，要格式化名称节点，操作步骤如下：

1、停止hdfs服务（执行stop-dfs.sh）
2、删除hadoop的临时目录tmp里的所有东西
3、格式化名称节点
首先，格式化配置HDFS文件系统，打开NameNode(HDFS服务器)，然后执行以下命令。 $ hadoop namenode -format
格式化HDFS后，启动分布式文件系统。以下命令将启动名称节点和数据节点的集群。 $ start-dfs.sh
4、重启hdfs服务（执行start-dfs.sh）

十、判断目录是否存在

创建DirExistsOrNot类：

package net.qing.hdfs;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.net.URI;

public class DirExistsOrNot {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String uri = "hdfs://hadoop129:9000";
        FileSystem fs = FileSystem.get(new URI(uri), conf);
        Path path = new Path("/park01");
        boolean result = fs.exists(path);
        if (result) {
            System.out.println("目录[" + path + "]存在！");
        } else {
            System.out.println("目录[" + path + "]不存在！");
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

HDFS Explorer查看

十一、判断Path指向的是目录还是文件

创建PathToFileOrDir类：

package net.qing.hdfs;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.net.URI;

public class PathToFileOrDir {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String uri = "hdfs://hadoop129:9000";
        FileSystem fs = FileSystem.get(new URI(uri), conf);
        Path path1 = new Path("/park02");
        if (fs.isDirectory(path1)) {
            System.out.println(path1 + "指向的是目录！");
        } else {
            System.out.println(path1 + "指向的是文件！");
        }

        Path path2 = new Path("/park02/text.doc");
        if (fs.isFile(path2)) {
            System.out.println(path2 + "指向的是文件！");
        } else {
            System.out.println(path2 + "指向的是目录！");
        }
    }
}
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

运行程序，结果如下：

十二、删除目录或文件

创建DeleteFileOrDir类：
（1）删除文件

任务：删除/park02/text.doc文件。

package net.hw.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.junit.Test;

import java.net.URI;

public class DeleteFileOrDir {
    @Test
    public void deleteFile() throws Exception {
        Configuration conf = new Configuration();
        String uri = "hdfs://hadoop129:9000";
        FileSystem fs = FileSystem.get(new URI(uri), conf);
        Path path = new Path("/park02/text.doc");
        boolean result = fs.delete(path, true);
        if (result) {
            System.out.println("文件[" + path + "]删除成功！");
        } else {
            System.out.println("文件[" + path + "]删除失败！");
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

再运行程序，结果如下：

可以在删除文件之前，判断文件是否存在。修改代码如下：

@Test                                                            
public void deleteFile() throws Exception {                      
    Configuration conf = new Configuration();                    
    String uri = "hdfs://hadoop129:9000";                            
    FileSystem fs = FileSystem.get(new URI(uri), conf);          
    Path path = new Path("/park02/text.txt");                   
    if (fs.exists(path)) {                                       
        boolean result = fs.delete(path, true);                  
        if (result) {                                            
            System.out.println("文件[" + path + "]删除成功！");         
        } else {                                                 
            System.out.println("文件[" + path + "]删除失败！");         
        }                                                        
    } else {                                                     
        System.out.println("文件[" + path + "]不存在！");              
    }                                                            
}                                                                
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

（2）删除目录

@Test
public void deleteDir() throws Exception {
    Configuration conf = new Configuration();
    String uri = "hdfs://hadoop129:9000";
    FileSystem fs = FileSystem.get(new URI(uri), conf);
    Path path = new Path("/park01");
    if (fs.exists(path)) {
        boolean result = fs.delete(path, true);
        if (result) {
            System.out.println("目录[" + path + "]删除成功！");
        } else {
            System.out.println("目录[" + path + "]删除失败！");
        }
    } else {
        System.out.println("目录[" + path + "]不存在！");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

再运行程序，结果如下：

（3）删除目录或文件
进行三个层面的判断：判断类型（目录或文件）、判断是否存在、判断删除是否成功。

@Test                                                                               
public void delete() throws Exception {                                             
    Configuration conf = new Configuration();                                       
    String uri = "hdfs://hadoop129:9000";                                               
    FileSystem fs = FileSystem.get(new URI(uri), conf);                             
    Path path1 = new Path("/park02/qjb.txt");                                       
    String type = "";                                                               
    if (fs.isFile(path1)) {                                                         
        type = "文件";                                                                
    } else {                                                                        
        type = "目录";                                                                
    }                                                                               
    if (fs.exists(path1)) {                                                         
        boolean result = fs.delete(path1, true);                                    
        if (result) {                                                               
            System.out.println(type + "[" + path1 + "]删除成功！");                      
        } else {                                                                    
            System.out.println(type + "[" + path1 + "]删除失败！");                      
        }                                                                           
    } else {                                                                        
        System.out.println(type + "[" + path1 + "]不存在！");                           
    }                                                                               
                                                                                    
    Path path2 = new Path("/park01");                                               
    if (fs.isFile(path1)) {                                                         
        type = "文件";                                                                
    } else {                                                                        
        type = "目录";                                                                
    }                                                                               
    if (fs.exists(path2)) {                                                         
        boolean result = fs.delete(path1, true);                                    
        if (result) {                                                               
            System.out.println(type + "[" + path2 + "]删除成功！");                      
        } else {                                                                    
            System.out.println(type + "[" + path2 + "]删除失败！");                      
        }                                                                           
    } else {                                                                        
        System.out.println(type + "[" + path2 + "]不存在！");                           
    }                                                                               
}                                                                                   
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

运行程序，结果如下: