Hadoop文件系统支持释疑之S3

Hadoop文件系统支持释疑之S3

一、引言

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.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 METADATA = 40 new HashMap(); 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 file197 closeQuietly(out);198 out = null; // to prevent a second close199 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 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 prefixes = new TreeSet();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 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 prefixes = new TreeSet();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 // ignore301 }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 }

View Code

四、有图有真相

五、结论

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

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。