Hadoop文件系统支持释疑之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 }