设为首页 加入收藏

TOP

Hadoop文件系统支持释疑之S3
2019-02-12 12:35:50 】 浏览:19
Tags:Hadoop 文件 系统 支持 释疑
版权声明:版权所有,作者:张子良,转载请注明出处。 https://blog.csdn.net/zhangziliang09/article/details/17794909

一、引言

  Hadoop版本提供了对多种文件系统的支持,但是这些文件系统是以何种方式实现的,其实现原理是什么以前并没有深究过。今天正好有人咨询我这个问题:Hadoop对S3的支持原理是什么?特此总结一下。Hadoop支持的文件系统包括:  

  文件系统 URI前缀 hadoop的具体实现类

  Local file fs.LocalFileSystem

  HDFS hdfs hdfs.DistributedFileSystem

  HFTP hftp hdfs.HftpFileSystem

  HSFTP hsftp hdfs.HsftpFileSystem

  HAR har fs.HarFileSystem

  KFS kfs fs.kfs.KosmosFileSystem

  FTP ftp fs.ftp.FTPFileSystem

  S3 (native) s3n fs.s3native.NativeS3FileSystem

  S3 (blockbased) s3 fs.s3.S3FileSystem

二、争议观点

  1.Hadoop对S3文件系统的支持是通过自己实现S3文件系统来做的吗?

   2.Hadoop对S3文件系统的支持是通过S3文件系统接口,实现的对S3文件系统的整合?

三、源码解析

复制代码
  1 package org.apache.hadoop.fs.s3;
  2 
  3 import java.io.BufferedInputStream;
  4 import java.io.BufferedOutputStream;
  5 import java.io.Closeable;
  6 import java.io.File;
  7 import java.io.FileInputStream;
  8 import java.io.FileOutputStream;
  9 import java.io.IOException;
 10 import java.io.InputStream;
 11 import java.io.OutputStream;
 12 import java.net.URI;
 13 import java.util.HashMap;
 14 import java.util.Map;
 15 import java.util.Set;
 16 import java.util.TreeSet;
 17 
 18 import org.apache.hadoop.conf.Configuration;
 19 import org.apache.hadoop.fs.Path;
 20 import org.apache.hadoop.fs.s3.INode.FileType;
 21 import org.jets3t.service.S3Service;
 22 import org.jets3t.service.S3ServiceException;
 23 import org.jets3t.service.impl.rest.httpclient.RestS3Service;
 24 import org.jets3t.service.model.S3Bucket;
 25 import org.jets3t.service.model.S3Object;
 26 import org.jets3t.service.security.AWSCredentials;
 27 
 28 class Jets3tFileSystemStore implements FileSystemStore {
 29   
 30   private static final String FILE_SYSTEM_NAME = "fs";
 31   private static final String FILE_SYSTEM_VALUE = "Hadoop";
 32 
 33   private static final String FILE_SYSTEM_TYPE_NAME = "fs-type";
 34   private static final String FILE_SYSTEM_TYPE_VALUE = "block";
 35 
 36   private static final String FILE_SYSTEM_VERSION_NAME = "fs-version";
 37   private static final String FILE_SYSTEM_VERSION_VALUE = "1";
 38   
 39   private static final Map<String, String> METADATA =
 40     new HashMap<String, String>();
 41   
 42   static {
 43     METADATA.put(FILE_SYSTEM_NAME, FILE_SYSTEM_VALUE);
 44     METADATA.put(FILE_SYSTEM_TYPE_NAME, FILE_SYSTEM_TYPE_VALUE);
 45     METADATA.put(FILE_SYSTEM_VERSION_NAME, FILE_SYSTEM_VERSION_VALUE);
 46   }
 47 
 48   private static final String PATH_DELIMITER = Path.SEPARATOR;
 49   private static final String BLOCK_PREFIX = "block_";
 50 
 51   private Configuration conf;
 52   
 53   private S3Service s3Service;
 54 
 55   private S3Bucket bucket;
 56   
 57   private int bufferSize;
 58   
 59   public void initialize(URI uri, Configuration conf) throws IOException {
 60     
 61     this.conf = conf;
 62     
 63     S3Credentials s3Credentials = new S3Credentials();
 64     s3Credentials.initialize(uri, conf);
 65     try {
 66       AWSCredentials awsCredentials =
 67         new AWSCredentials(s3Credentials.getAccessKey(),
 68             s3Credentials.getSecretAccessKey());
 69       this.s3Service = new RestS3Service(awsCredentials);
 70     } catch (S3ServiceException e) {
 71       if (e.getCause() instanceof IOException) {
 72         throw (IOException) e.getCause();
 73       }
 74       throw new S3Exception(e);
 75     }
 76     bucket = new S3Bucket(uri.getHost());
 77 
 78     this.bufferSize = conf.getInt("io.file.buffer.size", 4096);
 79   }
 80 
 81   public String getVersion() throws IOException {
 82     return FILE_SYSTEM_VERSION_VALUE;
 83   }
 84 
 85   private void delete(String key) throws IOException {
 86     try {
 87       s3Service.deleteObject(bucket, key);
 88     } catch (S3ServiceException e) {
 89       if (e.getCause() instanceof IOException) {
 90         throw (IOException) e.getCause();
 91       }
 92       throw new S3Exception(e);
 93     }
 94   }
 95 
 96   public void deleteINode(Path path) throws IOException {
 97     delete(pathToKey(path));
 98   }
 99 
100   public void deleteBlock(Block block) throws IOException {
101     delete(blockToKey(block));
102   }
103 
104   public boolean inodeExists(Path path) throws IOException {
105     InputStream in = get(pathToKey(path), true);
106     if (in == null) {
107       return false;
108     }
109     in.close();
110     return true;
111   }
112   
113   public boolean blockExists(long blockId) throws IOException {
114     InputStream in = get(blockToKey(blockId), false);
115     if (in == null) {
116       return false;
117     }
118     in.close();
119     return true;
120   }
121 
122   private InputStream get(String key, boolean checkMetadata)
123       throws IOException {
124     
125     try {
126       S3Object object = s3Service.getObject(bucket, key);
127       if (checkMetadata) {
128         checkMetadata(object);
129       }
130       return object.getDataInputStream();
131     } catch (S3ServiceException e) {
132       if ("NoSuchKey".equals(e.getS3ErrorCode())) {
133         return null;
134       }
135       if (e.getCause() instanceof IOException) {
136         throw (IOException) e.getCause();
137       }
138       throw new S3Exception(e);
139     }
140   }
141 
142   private InputStream get(String key, long byteRangeStart) throws IOException {
143     try {
144       S3Object object = s3Service.getObject(bucket, key, null, null, null,
145                                             null, byteRangeStart, null);
146       return object.getDataInputStream();
147     } catch (S3ServiceException e) {
148       if ("NoSuchKey".equals(e.getS3ErrorCode())) {
149         return null;
150       }
151       if (e.getCause() instanceof IOException) {
152         throw (IOException) e.getCause();
153       }
154       throw new S3Exception(e);
155     }
156   }
157 
158   private void checkMetadata(S3Object object) throws S3FileSystemException,
159       S3ServiceException {
160     
161     String name = (String) object.getMetadata(FILE_SYSTEM_NAME);
162     if (!FILE_SYSTEM_VALUE.equals(name)) {
163       throw new S3FileSystemException("Not a Hadoop S3 file.");
164     }
165     String type = (String) object.getMetadata(FILE_SYSTEM_TYPE_NAME);
166     if (!FILE_SYSTEM_TYPE_VALUE.equals(type)) {
167       throw new S3FileSystemException("Not a block file.");
168     }
169     String dataVersion = (String) object.getMetadata(FILE_SYSTEM_VERSION_NAME);
170     if (!FILE_SYSTEM_VERSION_VALUE.equals(dataVersion)) {
171       throw new VersionMismatchException(FILE_SYSTEM_VERSION_VALUE,
172           dataVersion);
173     }
174   }
175 
176   public INode retrieveINode(Path path) throws IOException {
177     return INode.deserialize(get(pathToKey(path), true));
178   }
179 
180   public File retrieveBlock(Block block, long byteRangeStart)
181     throws IOException {
182     File fileBlock = null;
183     InputStream in = null;
184     OutputStream out = null;
185     try {
186       fileBlock = newBackupFile();
187       in = get(blockToKey(block), byteRangeStart);
188       out = new BufferedOutputStream(new FileOutputStream(fileBlock));
189       byte[] buf = new byte[bufferSize];
190       int numRead;
191       while ((numRead = in.read(buf)) >= 0) {
192         out.write(buf, 0, numRead);
193       }
194       return fileBlock;
195     } catch (IOException e) {
196       // close output stream to file then delete file
197       closeQuietly(out);
198       out = null; // to prevent a second close
199       if (fileBlock != null) {
200         fileBlock.delete();
201       }
202       throw e;
203     } finally {
204       closeQuietly(out);
205       closeQuietly(in);
206     }
207   }
208   
209   private File newBackupFile() throws IOException {
210     File dir = new File(conf.get("fs.s3.buffer.dir"));
211     if (!dir.exists() && !dir.mkdirs()) {
212       throw new IOException("Cannot create S3 buffer directory: " + dir);
213     }
214     File result = File.createTempFile("input-", ".tmp", dir);
215     result.deleteOnExit();
216     return result;
217   }
218 
219   public Set<Path> listSubPaths(Path path) throws IOException {
220     try {
221       String prefix = pathToKey(path);
222       if (!prefix.endsWith(PATH_DELIMITER)) {
223         prefix += PATH_DELIMITER;
224       }
225       S3Object[] objects = s3Service.listObjects(bucket, prefix, PATH_DELIMITER);
226       Set<Path> prefixes = new TreeSet<Path>();
227       for (int i = 0; i < objects.length; i++) {
228         prefixes.add(keyToPath(objects[i].getKey()));
229       }
230       prefixes.remove(path);
231       return prefixes;
232     } catch (S3ServiceException e) {
233       if (e.getCause() instanceof IOException) {
234         throw (IOException) e.getCause();
235       }
236       throw new S3Exception(e);
237     }
238   }
239   
240   public Set<Path> listDeepSubPaths(Path path) throws IOException {
241     try {
242       String prefix = pathToKey(path);
243       if (!prefix.endsWith(PATH_DELIMITER)) {
244         prefix += PATH_DELIMITER;
245       }
246       S3Object[] objects = s3Service.listObjects(bucket, prefix, null);
247       Set<Path> prefixes = new TreeSet<Path>();
248       for (int i = 0; i < objects.length; i++) {
249         prefixes.add(keyToPath(objects[i].getKey()));
250       }
251       prefixes.remove(path);
252       return prefixes;
253     } catch (S3ServiceException e) {
254       if (e.getCause() instanceof IOException) {
255         throw (IOException) e.getCause();
256       }
257       throw new S3Exception(e);
258     }    
259   }
260 
261   private void put(String key, InputStream in, long length, boolean storeMetadata)
262       throws IOException {
263     
264     try {
265       S3Object object = new S3Object(key);
266       object.setDataInputStream(in);
267       object.setContentType("binary/octet-stream");
268       object.setContentLength(length);
269       if (storeMetadata) {
270         object.addAllMetadata(METADATA);
271       }
272       s3Service.putObject(bucket, object);
273     } catch (S3ServiceException e) {
274       if (e.getCause() instanceof IOException) {
275         throw (IOException) e.getCause();
276       }
277       throw new S3Exception(e);
278     }
279   }
280 
281   public void storeINode(Path path, INode inode) throws IOException {
282     put(pathToKey(path), inode.serialize(), inode.getSerializedLength(), true);
283   }
284 
285   public void storeBlock(Block block, File file) throws IOException {
286     BufferedInputStream in = null;
287     try {
288       in = new BufferedInputStream(new FileInputStream(file));
289       put(blockToKey(block), in, block.getLength(), false);
290     } finally {
291       closeQuietly(in);
292     }    
293   }
294 
295   private void closeQuietly(Closeable closeable) {
296     if (closeable != null) {
297       try {
298         closeable.close();
299       } catch (IOException e) {
300         // ignore
301       }
302     }
303   }
304 
305   private String pathToKey(Path path) {
306     if (!path.isAbsolute()) {
307       throw new IllegalArgumentException("Path must be absolute: " + path);
308     }
309     return path.toUri().getPath();
310   }
311 
312   private Path keyToPath(String key) {
313     return new Path(key);
314   }
315   
316   private String blockToKey(long blockId) {
317     return BLOCK_PREFIX + blockId;
318   }
319 
320   private String blockToKey(Block block) {
321     return blockToKey(block.getId());
322   }
323 
324   public void purge() throws IOException {
325     try {
326       S3Object[] objects = s3Service.listObjects(bucket);
327       for (int i = 0; i < objects.length; i++) {
328         s3Service.deleteObject(bucket, objects[i].getKey());
329       }
330     } catch (S3ServiceException e) {
331       if (e.getCause() instanceof IOException) {
332         throw (IOException) e.getCause();
333       }
334       throw new S3Exception(e);
335     }
336   }
337 
338   public void dump() throws IOException {
339     StringBuilder sb = new StringBuilder("S3 Filesystem, ");
340     sb.append(bucket.getName()).append("\n");
341     try {
342       S3Object[] objects = s3Service.listObjects(bucket, PATH_DELIMITER, null);
343       for (int i = 0; i < objects.length; i++) {
344         Path path = keyToPath(objects[i].getKey());
345         sb.append(path).append("\n");
346         INode m = retrieveINode(path);
347         sb.append("\t").append(m.getFileType()).append("\n");
348         if (m.getFileType() == FileType.DIRECTORY) {
349           continue;
350         }
351         for (int j = 0; j < m.getBlocks().length; j++) {
352           sb.append("\t").append(m.getBlocks()[j]).append("\n");
353         }
354       }
355     } catch (S3ServiceException e) {
356       if (e.getCause() instanceof IOException) {
357         throw (IOException) e.getCause();
358       }
359       throw new S3Exception(e);
360     }
361     System.out.println(sb);
362   }
363 
364 }
复制代码

四、有图有真相

五、结论

  Hadoop对S3文件系统的支持通过S3文件系统接口,实现的对S3文件系统的整合。有感兴趣的可以自行参照源码。


编程开发网
】【打印繁体】【投稿】【收藏】 【推荐】【举报】【评论】 【关闭】 【返回顶部
上一篇Mac配置单机版:Hadoop和Spark环境 下一篇CentOS7.5下搭建Hadoop2.9.1完全..

评论

帐  号: 密码: (新用户注册)
验 证 码:
表  情:
内  容:

array(4) { ["type"]=> int(8) ["message"]=> string(24) "Undefined variable: jobs" ["file"]=> string(32) "/mnt/wp/cppentry/do/bencandy.php" ["line"]=> int(214) }