title: 调用JAVA API 对 HDFS 进行文件的读取、写入、上传、下载、删除等操作
categories: java HADOOP
tags:

• java HADOOP
  abbrlink: 18170
  date: 2018-02-22 00:00:00

[ 调用JAVA API 对 HDFS 进行文件的读取、写入、上传、下载、删除等操作

](/qq_37279279/article/details/79291461)

Hadoop文件系统
基本的文件系统命令操作, 通过hadoop fs -help可以获取所有的命令的详细帮助文件。

Java抽象类org.apache.hadoop.fs.FileSystem定义了hadoop的一个文件系统接口。该类是一个抽象类，通过以下两种静态工厂方法可以过去FileSystem实例：
public static FileSystem.get(Configuration conf) throws IOException
public static FileSystem.get(URI uri, Configuration conf) throws IOException

具体方法实现：
1、public boolean mkdirs(Path f) throws IOException
一次性新建所有目录（包括父目录）， f是完整的目录路径。

2、public FSOutputStream create(Path f) throws IOException
创建指定path对象的一个文件，返回一个用于写入数据的输出流
create()有多个重载版本，允许我们指定是否强制覆盖已有的文件、文件备份数量、写入文件缓冲区大小、文件块大小以及文件权限。

3、public boolean copyFromLocal(Path src, Path dst) throws IOException
将本地文件拷贝到文件系统

4、public boolean exists(Path f) throws IOException
检查文件或目录是否存在

5、public boolean delete(Path f, Boolean recursive)
永久性删除指定的文件或目录，如果f是一个空目录或者文件，那么recursive的值就会被忽略。只有recursive＝true时，一个非空目录及其内容才会被删除。

6、FileStatus类封装了文件系统中文件和目录的元数据，包括文件长度、块大小、备份、修改时间、所有者以及权限信息。

[java] view plain copy

1. import org.apache.hadoop.conf.Configuration;
2. import org.apache.hadoop.fs.*;
3. import org.apache.hadoop.fs.FileSystem;
4. import org.apache.hadoop.io.IOUtils;
5. 6. import java.io.*;
6. import java.net.URI;
7. import java.net.URISyntaxException;
8. import java.util.ArrayList;
9. import java.util.List;
10. 12. public class HdfsOperate {
11. ArrayList hdfsfiles;
12. 15. public HdfsOperate() {
13. this .hdfsfiles = new ArrayList();
14. }
15. 19. /**
16. • 获取hdfs路径下的文件列表
17. *
18. • @param srcpath
19. • @return
20. */
21. public String[] getFileList(String srcpath) {
22. try {
23. Configuration conf = new Configuration();
24. Path path = new Path(srcpath);
25. FileSystem fs = path.getFileSystem(conf);
26. List files = new ArrayList();
27. if (fs.exists(path) && fs.isDirectory(path)) {
28. for (FileStatus status : fs.listStatus(path)) {
29. files.add(status.getPath().toString());
30. }
31. }
32. //fs.close();
33. return files.toArray( new String[]{});
34. } catch (IOException e) {
35. } catch (Exception e) {
36. }
37. return null ;
38. }
39. 44. /**
40. • 给定文件名和文件内容，创建hdfs文件
41. *
42. • @param dst
43. • @param contents
44. • @throws IOException
45. */
46. public void createFile(String dst, byte [] contents)
47. throws IOException {
48. Configuration conf = new Configuration();
49. Path dstPath = new Path(dst);
50. FileSystem fs = dstPath.getFileSystem(conf);
51. 57. FSDataOutputStream outputStream = fs.create(dstPath);
52. outputStream.write(contents);
53. outputStream.close();
54. System.out.println( “create file “ + dst + “ success!” );
55. //fs.close();
56. }
57. 64. /**
58. • 删除hdfs文件
59. *
60. • @param filePath
61. • @throws IOException
62. */
63. public void delete(String filePath) throws IOException {
64. Configuration conf = new Configuration();
65. Path path = new Path(filePath);
66. FileSystem fs = path.getFileSystem(conf);
67. 75. boolean isok = fs.deleteOnExit(path);
68. if (isok) {
69. System.out.println( “delete file “ + filePath + “ success!” );
70. } else {
71. System.out.println( “delete file “ + filePath + “ failure” );
72. }
73. //fs.close();
74. }
75. 84. /**
76. • 创建hdfs目录
77. *
78. • @param path
79. • @throws IOException
80. */
81. public void mkdir(String path) throws IOException {
82. Configuration conf = new Configuration();
83. Path srcPath = new Path(path);
84. FileSystem fs = srcPath.getFileSystem(conf);
85. 95. boolean isok = fs.mkdirs(srcPath);
86. if (isok) {
87. System.out.println( “create dir ok!” );
88. } else {
89. System.out.println( “create dir failure” );
90. }
91. //fs.close();
92. }
93. 104. /**
94. • 读取hdfs文件内容，并在控制台打印出来
95. *
96. • @param filePath
97. • @throws IOException
98. */
99. public void readFile(String filePath) throws IOException {
100. Configuration conf = new Configuration();
101. Path srcPath = new Path(filePath);
102. FileSystem fs = null ;
103. URI uri;
104. try {
105. uri = new URI(filePath);
106. fs = FileSystem.get(uri, conf);
107. } catch (URISyntaxException e) {
108. e.printStackTrace();
109. }
110. InputStream in = null ;
111. try {
112. in = fs.open(srcPath);
113. IOUtils.copyBytes(in, System.out, 4096 , false );
114. } finally {
115. IOUtils.closeStream(in);
116. }
117. }
118. 130. /**
119. • 下载hdfs文件到本地目录
120. *
121. • @param dstPath
122. • @param srcPath
123. • @throws Exception
124. */
125. public void downloadFile(String dstPath, String srcPath) throws Exception {
126. Path path = new Path(srcPath);
127. Configuration conf = new Configuration();
128. FileSystem hdfs = path.getFileSystem(conf);
129. 142. File rootfile = new File(dstPath);
130. if (!rootfile.exists()) {
131. rootfile.mkdirs();
132. }
133. 147. if (hdfs.isFile(path)) {
134. //只下载非txt文件
135. String fileName = path.getName();
136. if (!fileName.toLowerCase().endsWith( “txt” )) {
137. FSDataInputStream in = null ;
138. FileOutputStream out = null ;
139. try {
140. in = hdfs.open(path);
141. File srcfile = new File(rootfile, path.getName());
142. if (!srcfile.exists()) srcfile.createNewFile();
143. out = new FileOutputStream(srcfile);
144. IOUtils.copyBytes(in, out, 4096 , false );
145. } finally {
146. IOUtils.closeStream(in);
147. IOUtils.closeStream(out);
148. }
149. //下载完后，在hdfs上将原文件删除
150. this .delete(path.toString());
151. }
152. } else if (hdfs.isDirectory(path)) {
153. File dstDir = new File(dstPath);
154. if (!dstDir.exists()) {
155. dstDir.mkdirs();
156. }
157. //在本地目录上加一层子目录
158. String filePath = path.toString(); //目录
159. String subPath[] = filePath.split( “/“ );
160. String newdstPath = dstPath + subPath[subPath.length - 1 ] + “/“ ;
161. System.out.println( “newdstPath=======” + newdstPath);
162. if (hdfs.exists(path) && hdfs.isDirectory(path)) {
163. FileStatus[] srcFileStatus = hdfs.listStatus(path);
164. if (srcFileStatus != null ) {
165. for (FileStatus status : hdfs.listStatus(path)) {
166. //下载子目录下文件
167. downloadFile(newdstPath, status.getPath().toString());
168. }
169. }
170. }
171. }
172. }
173. 188. /**
174. • 下载hdfs文件内容，保存到内存对象中
175. *
176. • @param srcPath
177. • @throws Exception
178. */
179. public void downloadFileByte(String srcPath) throws Exception {
180. Path path = new Path(srcPath);
181. FileSystem hdfs = null ;
182. Configuration conf = new Configuration();
183. hdfs = FileSystem.get(URI.create(srcPath), conf);
184. if (hdfs.exists(path)) {
185. if (hdfs.isFile(path)) {
186. //如果是文件
187. FSDataInputStream in = null ;
188. FileOutputStream out = null ;
189. try {
190. in = hdfs.open( new Path(srcPath));
191. byte [] t = new byte [in.available()];
192. in.read(t);
193. hdfsfiles.add( new HdfsFile(path.getName(), srcPath, t));
194. } finally {
195. IOUtils.closeStream(in);
196. IOUtils.closeStream(out);
197. }
198. } else {
199. //如果是目录
200. FileStatus[] srcFileStatus = hdfs.listStatus( new Path(srcPath));
201. for ( int i = 0 ; i < srcFileStatus.length; i++) {
202. String srcFile = srcFileStatus[i].getPath().toString();
203. downloadFileByte(srcFile);
204. }
205. }
206. }
207. }
208. 224. public ArrayList getHdfsfiles() {
209. return hdfsfiles;
210. }
211. 228. /**
212. • 将本地目录或文件上传的hdfs
213. *
214. • @param localSrc
215. • @param dst
216. • @throws Exception
217. */
218. public void uploadFile(String localSrc, String dst) throws Exception {
219. 237. Configuration conf = new Configuration();
220. File srcFile = new File(localSrc);
221. if (srcFile.isDirectory()) {
222. copyDirectory(localSrc, dst, conf);
223. } else {
224. copyFile(localSrc, dst, conf);
225. }
226. }
227. 246. /**
228. • 拷贝本地文件hdfs目录下
229. *
230. • @param localSrc
231. • @param dst
232. • @param conf
233. • @return
234. • @throws Exception
235. */
236. private boolean copyFile(String localSrc, String dst, Configuration conf) throws Exception {
237. File file = new File(localSrc);
238. dst = dst + file.getName();
239. Path path = new Path(dst);
240. FileSystem fs = path.getFileSystem(conf); //FileSystem.get(conf);
241. fs.exists(path);
242. InputStream in = new BufferedInputStream( new FileInputStream(file));
243. OutputStream out = fs.create( new Path(dst));
244. IOUtils.copyBytes(in, out, 4096 , true );
245. in.close();
246. return true ;
247. }
248. 268. /**
249. • 拷贝本地目录到hdfs
250. • @param src
251. • @param dst
252. • @param conf
253. • @return
254. • @throws Exception
255. */
256. private boolean copyDirectory(String src, String dst, Configuration conf) throws Exception {
257. Path path = new Path(dst);
258. FileSystem fs = path.getFileSystem(conf);
259. if (!fs.exists(path)) {
260. fs.mkdirs(path);
261. }
262. File file = new File(src);
263. 284. File[] files = file.listFiles();
264. for ( int i = 0 ; i < files.length; i++) {
265. File f = files[i];
266. if (f.isDirectory()) {
267. String fname = f.getName();
268. if (dst.endsWith( “/“ )) {
269. copyDirectory(f.getPath(), dst + fname + “/“ , conf);
270. } else {
271. copyDirectory(f.getPath(), dst + “/“ + fname + “/“ , conf);
272. }
273. } else {
274. copyFile(f.getPath(), dst, conf);
275. }
276. }
277. return true ;
278. }
279. }