Java API操作HDFS

Posted 2020-06-12

HDFS是存储数据的分布式文件系统，对HDFS的操作，就是对文件系统的操作，除了用HDFS的shell命令对文件系统进行操作，我们也可以利用Java API对文件系统进行操作，比如文件的创建、删除、修改权限等等，还有文件夹的创建、删除、重命名等等。

使用Java API对文件系统进行操作主要涉及以下几个类：

1.Configuration类：该类的对象封装了客户端或者服务端的配置。

2.FileSystem类：该类的对象是一个文件系统对象，可以利用该对象的一些方法来对文件进行操作，FileSystem是一个抽象，不能通过new来获取对象，应该通过Filesystem的静态方法get()来获取：

 

[java] view plain copy

 

 

1. //创建URI对象  
2.             //PATH = "hdfs://liaozhongmin:9000/";  
3.             URI uri = new URI(PATH);  
4.             //获取文件系统  
5.             FileSystem fileSystem = FileSystem.get(uri, new Configuration());  

3.FSDataInputStream和FSDataOutputStream：这两个类是HDFS中的输入输出流。分别通过FileSystem的open()方法和create()方法获得。

 

 

在使用Java API 操作HDFS之前，首先要保证hadoop是正常启动的，可以通过jps命令来查看hadoop的进程是否全部启动，如下图：

注：如上图所示，使用jps命令可以看到Hadoop的五个进程即NameNode、DataNode、SecondaryNameNode、TaskTrackers、JobTracker启动了，就表示hadoop启动成功了。

 

下面是关于Java API操作HDFS文件系统的常用工具类：

 

[java] view plain copy

 

 

1. public class FileSystemUtil {  
2.   
3.     //定义HDFS的路径  
4.     private static final String PATH = "hdfs://liaozhongmin:9000/";  
5.       
6.     /** 
7.      * 获取文件系统FileSystem 
8.      * @return 
9.      */  
10.     public static FileSystem getFileSystem(){  
11.           
12.         try {  
13.             //创建URI对象  
14.             //PATH = "hdfs://liaozhongmin:9000/";  
15.             URI uri = new URI(PATH);  
16.             //获取文件系统  
17.             FileSystem fileSystem = FileSystem.get(uri, new Configuration());  
18.             return fileSystem;  
19.         } catch (Exception e) {  
20.             e.printStackTrace();  
21.         }  
22.           
23.         return null;  
24.     }  
25.       
26.     /** 
27.      * 创建文件夹 
28.      * @param fileSystem 
29.      */  
30.     public static void mkdir(FileSystem fileSystem,String path){  
31.           
32.         try {  
33.             //创建Path对象  
34.             Path srcPath = new Path(path);  
35.             //通过文件系统对象创建文件夹  
36.             fileSystem.mkdirs(srcPath);  
37.         } catch (IOException e) {  
38.             e.printStackTrace();  
39.         } finally{  
40.             try {  
41.                 fileSystem.close();  
42.             } catch (IOException e) {  
43.                 e.printStackTrace();  
44.             }  
45.         }  
46.     }  
47.       
48.     /** 
49.      * 创建新文件 
50.      * @param dst 
51.      * @param contents 
52.      */  
53.     public static void createFile(String dst,byte[] contents){  
54.           
55.         //定义文件系统  
56.         FileSystem fileSystem = null;  
57.         //定义输出流  
58.         FSDataOutputStream outputStream = null;  
59.         try {  
60.             //获取文件系统  
61.             fileSystem = FileSystemUtil.getFileSystem();  
62.             //创建一个目标路径  
63.             Path path = new Path(dst);  
64.             //创建文件  
65.             outputStream = fileSystem.create(path);  
66.             //写数据  
67.             outputStream.write(contents);  
68.             System.out.println("文件创建成功!");  
69.         } catch (IOException e) {  
70.             e.printStackTrace();  
71.         } finally{  
72.             //关闭  
73.             try {  
74.                 fileSystem.close();  
75.                 outputStream.close();  
76.             } catch (IOException e) {  
77.                 e.printStackTrace();  
78.             }  
79.               
80.         }  
81.           
82.     }  
83.     /** 
84.      * 向文件系统中上传本地数据 
85.      * @param fileSystem 
86.      */  
87.     public static void putData(FileSystem fileSystem,String src,String dst){  
88.           
89.         try {  
90.             //创建原路径Path对象  
91.             Path srcPath = new Path(src);  
92.             //创建目标路径Path对象  
93.             Path dstPath = new Path(dst);  
94.             //调用文件系统的文件爱你复制函数，前面的参数是指是否删除源文件，true为删除，否则不删除  
95.             fileSystem.copyFromLocalFile(false, srcPath, dstPath);  
96.               
97.         } catch (IOException e) {  
98.             e.printStackTrace();  
99.         } finally{  
100.             //关闭文件系统  
101.             try {  
102.                 fileSystem.close();  
103.             } catch (IOException e) {  
104.                 e.printStackTrace();  
105.             }  
106.         }  
107.     }  
108.       
109.     /** 
110.      * 文件重命名 
111.      * @param oldName 旧名字 
112.      * @param newName 新名字 
113.      */  
114.     public static void rename(FileSystem fileSystem,String oldName,String newName){  
115.           
116.         try {  
117.             //创建旧文件的Path对象  
118.             Path oldPath = new Path(oldName);  
119.             Path newPath = new Path(newName);  
120.             //重命名  
121.             fileSystem.rename(oldPath, newPath);  
122.         } catch (Exception e) {  
123.             e.printStackTrace();  
124.         } finally{  
125.             try {  
126.                 fileSystem.close();  
127.             } catch (IOException e) {  
128.                 e.printStackTrace();  
129.             }  
130.         }  
131.     }  
132.     /** 
133.      * 下载文件 
134.      * @param fileSystem 
135.      * @param fileName 
136.      */  
137.     public static void getData(FileSystem fileSystem,String src,String dst){  
138.           
139.           
140.         try {  
141.             //创建Path对象  
142.             Path path = new Path(src);  
143.             //通过path构建文件系统的输入流  
144.             FSDataInputStream in = fileSystem.open(path);  
145.             //构建文件  
146.             File file = new File(dst);  
147.               
148.             if (!file.exists()){  
149.                 file.createNewFile();  
150.             }  
151.             //构建输出流  
152.             FileOutputStream fileOutputStream = new FileOutputStream(file);  
153.             //下载  
154.             IOUtils.copyBytes(in, fileOutputStream, 1024, true);  
155.         } catch (IOException e) {  
156.               
157.             e.printStackTrace();  
158.         }  
159.     }  
160.       
161.     /** 
162.      * 遍历文件系统中的某个目录 
163.      * @param fileSystem 
164.      */  
165.     public static void listFile(FileSystem fileSystem,String path){  
166.           
167.         try {  
168.             //调用listStatus()方法获取一个文件数组  
169.             FileStatus[] listStatus = fileSystem.listStatus(new Path(path));  
170.             //循环遍历  
171.             for (FileStatus fileStatus : listStatus){  
172.                 //判断是否为目录  
173.                 String isDir = fileStatus.isDir()?"是文件夹":"是文件";  
174.                 //获取文件的权限  
175.                 String permission = fileStatus.getPermission().toString();  
176.                 //获取备份  
177.                 short replication = fileStatus.getReplication();  
178.                 //获取数组的长度  
179.                 long len = fileStatus.getLen();  
180.                 //获取文件的路径  
181.                 String filePath = fileStatus.getPath().toString();  
182.                 //打印数据  
183.                 System.out.println("isDir:" +isDir + "\npermission:" + permission + "\nreplication:" + replication+ "\nlen:" + len + "\nfilepath:" + filePath);  
184.             }  
185.         } catch (IOException e) {  
186.               
187.             e.printStackTrace();  
188.         }  
189.     }  
190.       
191.     /** 
192.      * 删除文件 
193.      * @param fileSystem 
194.      */  
195.     public static void remove(FileSystem fileSystem,String filePath){  
196.           
197.         try {  
198.             //创建path对象  
199.             Path path = new Path(filePath);  
200.             //通过文件系统进行删除,第二个参数设置为true时才可以删除为目录的文件，否则的话如果是目录而不是文件就会报错的  
201.             fileSystem.delete(path, true);  
202.         } catch (IOException e) {  
203.             e.printStackTrace();  
204.         }  
205.     }  
206.       
207.     /** 
208.      * 工具类的测试 
209.      * @param args 
210.      */  
211.     public static void main(String[] args) {  
212.           
213.         //创建文件系统  
214.         FileSystem fileSystem = getFileSystem();  
215.           
216.         //创建文件夹  
217.         //FileSystemUtil.mkdir(fileSystem, "/myDir");  
218.           
219.         //创建文件并写入内容  
220.         //byte[] contents = "我爱你".getBytes();  
221.         //FileSystemUtil.createFile("/myDir/liao.txt", contents);  
222.           
223.         //文件重命名  
224.         //FileSystemUtil.rename(fileSystem, "/myDir/liao.txt", "/myDir/liao_bak");  
225.           
226.         //上传文件  
227.         //FileSystemUtil.putData(fileSystem, "D:\\android\\test.txt", "/myDir");  
228.           
229.         //下载文件  
230.         //FileSystemUtil.getData(fileSystem, "/myDir/test.txt", "D:\\test.txt");  
231.           
232.         //遍历文件系统的某个目录  
233.         //FileSystemUtil.listFile(fileSystem, "/");  
234.           
235.         //删除文件  
236.         FileSystemUtil.remove(fileSystem, "/dir/liaozhongmin.txt");  
237.     }  
238. }  

 

 

注：远程文件系统的路径，请自行在hadoop/conf/core-site.xml文件中配置：

[html] view plain copy

 

 

1. <?xml version="1.0"?>  
2. <?xml-stylesheet type="text/xsl" href="configuration.xsl"?>  
3.   
4. <!-- Put site-specific property overrides in this file. -->  
5.   
6. <configuration>  
7.      <property>  
8.              <name>fs.default.name</name>  
9.         <!--文件系统的路径-->  
10.              <value>hdfs://liaozhongmin:9000</value>  
11.       </property>  
12.   
13.       <property>  
14.               <name>hadoop.tmp.dir</name>  
15.          <!--文件保存的位置(上传后以块的形式保存)-->  
16.               <value>/usr/local/hadoop/tmp</value>  
17.       </property>    
18.   
19. </configuration>