安科网

Hadoop文件系统支持释疑之S3

心中永远的正能量

心中永远的正能量

2014-01-03

关注 关注

一、引言
　　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文件系统的整合？

三、源码解析

package org.apache.hadoop.fs.s3;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.Closeable;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.s3.INode.FileType;
import org.jets3t.service.S3Service;
import org.jets3t.service.S3ServiceException;
import org.jets3t.service.impl.rest.httpclient.RestS3Service;
import org.jets3t.service.model.S3Bucket;
import org.jets3t.service.model.S3Object;
import org.jets3t.service.security.AWSCredentials;

class Jets3tFileSystemStore implements FileSystemStore {
 
  private static final String FILE_SYSTEM_NAME = "fs";
  private static final String FILE_SYSTEM_VALUE = "Hadoop";

  private static final String FILE_SYSTEM_TYPE_NAME = "fs-type";
  private static final String FILE_SYSTEM_TYPE_VALUE = "block";

  private static final String FILE_SYSTEM_VERSION_NAME = "fs-version";
  private static final String FILE_SYSTEM_VERSION_VALUE = "1";
 
  private static final Map<String, String> METADATA =
    new HashMap<String, String>();
 
  static {
    METADATA.put(FILE_SYSTEM_NAME, FILE_SYSTEM_VALUE);
    METADATA.put(FILE_SYSTEM_TYPE_NAME, FILE_SYSTEM_TYPE_VALUE);
    METADATA.put(FILE_SYSTEM_VERSION_NAME, FILE_SYSTEM_VERSION_VALUE);
  }

  private static final String PATH_DELIMITER = Path.SEPARATOR;
  private static final String BLOCK_PREFIX = "block_";

  private Configuration conf;
 
  private S3Service s3Service;

  private S3Bucket bucket;
 
  private int bufferSize;
 
  public void initialize(URI uri, Configuration conf) throws IOException {
   
    this.conf = conf;
   
    S3Credentials s3Credentials = new S3Credentials();
    s3Credentials.initialize(uri, conf);
    try {
      AWSCredentials awsCredentials =
        new AWSCredentials(s3Credentials.getAccessKey(),
            s3Credentials.getSecretAccessKey());
      this.s3Service = new RestS3Service(awsCredentials);
    } catch (S3ServiceException e) {
      if (e.getCause() instanceof IOException) {
        throw (IOException) e.getCause();
      }
      throw new S3Exception(e);
    }
    bucket = new S3Bucket(uri.getHost());

    this.bufferSize = conf.getInt("io.file.buffer.size", 4096);
  }

  public String getVersion() throws IOException {
    return FILE_SYSTEM_VERSION_VALUE;
  }

  private void delete(String key) throws IOException {
    try {
      s3Service.deleteObject(bucket, key);
    } catch (S3ServiceException e) {
      if (e.getCause() instanceof IOException) {
        throw (IOException) e.getCause();
      }
      throw new S3Exception(e);
    }
  }

  public void deleteINode(Path path) throws IOException {
    delete(pathToKey(path));
  }

  public void deleteBlock(Block block) throws IOException {
    delete(blockToKey(block));
  }

  public boolean inodeExists(Path path) throws IOException {
    InputStream in = get(pathToKey(path), true);
    if (in == null) {
      return false;
    }
    in.close();
    return true;
  }
 
  public boolean blockExists(long blockId) throws IOException {
    InputStream in = get(blockToKey(blockId), false);
    if (in == null) {
      return false;
    }
    in.close();
    return true;
  }

  private InputStream get(String key, boolean checkMetadata)
      throws IOException {
   
    try {
      S3Object object = s3Service.getObject(bucket, key);
      if (checkMetadata) {
        checkMetadata(object);
      }
      return object.getDataInputStream();
    } catch (S3ServiceException e) {
      if ("NoSuchKey".equals(e.getS3ErrorCode())) {
        return null;
      }
      if (e.getCause() instanceof IOException) {
        throw (IOException) e.getCause();
      }
      throw new S3Exception(e);
    }
  }

  private InputStream get(String key, long byteRangeStart) throws IOException {
    try {
      S3Object object = s3Service.getObject(bucket, key, null, null, null,
                                            null, byteRangeStart, null);
      return object.getDataInputStream();
    } catch (S3ServiceException e) {
      if ("NoSuchKey".equals(e.getS3ErrorCode())) {
        return null;
      }
      if (e.getCause() instanceof IOException) {
        throw (IOException) e.getCause();
      }
      throw new S3Exception(e);
    }
  }

  private void checkMetadata(S3Object object) throws S3FileSystemException,
      S3ServiceException {
   
    String name = (String) object.getMetadata(FILE_SYSTEM_NAME);
    if (!FILE_SYSTEM_VALUE.equals(name)) {
      throw new S3FileSystemException("Not a Hadoop S3 file.");
    }
    String type = (String) object.getMetadata(FILE_SYSTEM_TYPE_NAME);
    if (!FILE_SYSTEM_TYPE_VALUE.equals(type)) {
      throw new S3FileSystemException("Not a block file.");
    }
    String dataVersion = (String) object.getMetadata(FILE_SYSTEM_VERSION_NAME);
    if (!FILE_SYSTEM_VERSION_VALUE.equals(dataVersion)) {
      throw new VersionMismatchException(FILE_SYSTEM_VERSION_VALUE,
          dataVersion);
    }
  }

  public INode retrieveINode(Path path) throws IOException {
    return INode.deserialize(get(pathToKey(path), true));
  }

  public File retrieveBlock(Block block, long byteRangeStart)
    throws IOException {
    File fileBlock = null;
    InputStream in = null;
    OutputStream out = null;
    try {
      fileBlock = newBackupFile();
      in = get(blockToKey(block), byteRangeStart);
      out = new BufferedOutputStream(new FileOutputStream(fileBlock));
      byte[] buf = new byte[bufferSize];
      int numRead;
      while ((numRead = in.read(buf)) >= 0) {
        out.write(buf, 0, numRead);
      }
      return fileBlock;
    } catch (IOException e) {
      // close output stream to file then delete file
      closeQuietly(out);
      out = null; // to prevent a second close
      if (fileBlock != null) {
        fileBlock.delete();
      }
      throw e;
    } finally {
      closeQuietly(out);
      closeQuietly(in);
    }
  }
 
  private File newBackupFile() throws IOException {
    File dir = new File(conf.get("fs.s3.buffer.dir"));
    if (!dir.exists() && !dir.mkdirs()) {
      throw new IOException("Cannot create S3 buffer directory: " + dir);
    }
    File result = File.createTempFile("input-", ".tmp", dir);
    result.deleteOnExit();
    return result;
  }

  public Set<Path> listSubPaths(Path path) throws IOException {
    try {
      String prefix = pathToKey(path);
      if (!prefix.endsWith(PATH_DELIMITER)) {
        prefix += PATH_DELIMITER;
      }
      S3Object[] objects = s3Service.listObjects(bucket, prefix, PATH_DELIMITER);
      Set<Path> prefixes = new TreeSet<Path>();
      for (int i = 0; i < objects.length; i++) {
        prefixes.add(keyToPath(objects[i].getKey()));
      }
      prefixes.remove(path);
      return prefixes;
    } catch (S3ServiceException e) {
      if (e.getCause() instanceof IOException) {
        throw (IOException) e.getCause();
      }
      throw new S3Exception(e);
    }
  }
 
  public Set<Path> listDeepSubPaths(Path path) throws IOException {
    try {
      String prefix = pathToKey(path);
      if (!prefix.endsWith(PATH_DELIMITER)) {
        prefix += PATH_DELIMITER;
      }
      S3Object[] objects = s3Service.listObjects(bucket, prefix, null);
      Set<Path> prefixes = new TreeSet<Path>();
      for (int i = 0; i < objects.length; i++) {
        prefixes.add(keyToPath(objects[i].getKey()));
      }
      prefixes.remove(path);
      return prefixes;
    } catch (S3ServiceException e) {
      if (e.getCause() instanceof IOException) {
        throw (IOException) e.getCause();
      }
      throw new S3Exception(e);
    }   
  }

  private void put(String key, InputStream in, long length, boolean storeMetadata)
      throws IOException {
   
    try {
      S3Object object = new S3Object(key);
      object.setDataInputStream(in);
      object.setContentType("binary/octet-stream");
      object.setContentLength(length);
      if (storeMetadata) {
        object.addAllMetadata(METADATA);
      }
      s3Service.putObject(bucket, object);
    } catch (S3ServiceException e) {
      if (e.getCause() instanceof IOException) {
        throw (IOException) e.getCause();
      }
      throw new S3Exception(e);
    }
  }

  public void storeINode(Path path, INode inode) throws IOException {
    put(pathToKey(path), inode.serialize(), inode.getSerializedLength(), true);
  }

  public void storeBlock(Block block, File file) throws IOException {
    BufferedInputStream in = null;
    try {
      in = new BufferedInputStream(new FileInputStream(file));
      put(blockToKey(block), in, block.getLength(), false);
    } finally {
      closeQuietly(in);
    }   
  }

  private void closeQuietly(Closeable closeable) {
    if (closeable != null) {
      try {
        closeable.close();
      } catch (IOException e) {
        // ignore
      }
    }
  }

  private String pathToKey(Path path) {
    if (!path.isAbsolute()) {
      throw new IllegalArgumentException("Path must be absolute: " + path);
    }
    return path.toUri().getPath();
  }

  private Path keyToPath(String key) {
    return new Path(key);
  }
 
  private String blockToKey(long blockId) {
    return BLOCK_PREFIX + blockId;
  }

  private String blockToKey(Block block) {
    return blockToKey(block.getId());
  }

  public void purge() throws IOException {
    try {
      S3Object[] objects = s3Service.listObjects(bucket);
      for (int i = 0; i < objects.length; i++) {
        s3Service.deleteObject(bucket, objects[i].getKey());
      }
    } catch (S3ServiceException e) {
      if (e.getCause() instanceof IOException) {
        throw (IOException) e.getCause();
      }
      throw new S3Exception(e);
    }
  }

  public void dump() throws IOException {
    StringBuilder sb = new StringBuilder("S3 Filesystem, ");
    sb.append(bucket.getName()).append("\n");
    try {
      S3Object[] objects = s3Service.listObjects(bucket, PATH_DELIMITER, null);
      for (int i = 0; i < objects.length; i++) {
        Path path = keyToPath(objects[i].getKey());
        sb.append(path).append("\n");
        INode m = retrieveINode(path);
        sb.append("\t").append(m.getFileType()).append("\n");
        if (m.getFileType() == FileType.DIRECTORY) {
          continue;
        }
        for (int j = 0; j < m.getBlocks().length; j++) {
          sb.append("\t").append(m.getBlocks()[j]).append("\n");
        }
      }
    } catch (S3ServiceException e) {
      if (e.getCause() instanceof IOException) {
        throw (IOException) e.getCause();
      }
      throw new S3Exception(e);
    }
    System.out.println(sb);
  }

}

四、有图有真相

Hadoop文件系统支持释疑之S3

相关阅读：

s3 hadoop hdfs

心中永远的正能量

心中永远的正能量

0 关注 0 粉丝 0 动态

关注 关注

相关推荐

AWS CloudFront CDN + S3 CORS 跨域访问的问题

因为上面的限制，你需要调整 CloudFront 和 S3。选择行为选项卡。如果更新 CORS 策略并将相应的标头列入白名单后仍显示错误，请尝试在分配的缓存行为中允许 OPTIONS HTTP 方法。默认情况下，CloudFront 只允许 GET 和 H

wolfjin 2020-04-21

python字符串操作

The string whose method is called is inserted in between each given string.Example: ‘.‘.join -> ‘ab.pq.rs‘。print #立即返回一个三元

waiwaiLILI 2020-03-20

aws s3 上传 binary 数据 （通过stringstream）

有个需求需要将二进制istream上传到s3上暂存，但苦于没能直接找到接口，官方提供的设置数据块的接口如下：。在实际寻找传入参数的时候没找到比较好的传入对象：fstream意味着我需要先将数据存到磁盘再去读取，感觉不是很好；而stringstream在装二

kaylin0 2020-01-17

笨办法学AWS:各类存储服务适用场景总结

Amazon S3 is used to store and distribute static web content and media 静态网站内容和媒体文件。Amazon S3 is often used as a highly durable,

longshilincom 2020-01-14

笨办法学AWS:各类存储服务适用场景总结

Amazon S3 is used to store and distribute static web content and media 静态网站内容和媒体文件。Amazon S3 is often used as a highly durable,

JokerCch 2020-01-14

aws 的 s3 put_object vs upload_file

用过aws的人，都知道s3存储东西贼方便。第一个是get_object(),是将文件的内容赋值给body,进行上传，并设置存储桶为上传文件为公开。这种方法，就是上传的时候只需要填入文件的名称就可以实现上传，但是貌似不能设置这个文件的公开需要去配置s3的存储

黑夜流星 2019-12-26

Python之set

#set和dict类似，也是一组key的集合，但不存储value。由于key不能重复，所以，在set中，没有重复的key。#此时 传入的参数[1,2,3]是一个list,而显示的{1,2,3}是说明set内部有1,2,3这三个元素，显示的顺序不代表它是有序

georgeandgeorge 2019-12-20

使用Python和Django将文件上传到AWS S3

介绍为了建立具有更多互动性的网站，我们不仅要向用户传递信息，还要允许他们上传自己的数据。这为我们的网站提供了更多的服务终端用户的机会和方式。通过允许用户上传文件，我们可以允许他们与他人分享照片、视频或音乐，或者为了安全起见备份这些文件。社交媒体在全球范围内

beautiful00 2019-10-23

Python 设置S3文件属性中元数据的Content-Encoding值

背景: 使用脚本从S3下载下来的CSV文件打开发现是乱码,但是在网页上面点击下载按钮下载后却能正常打开.浏览器下载的时候会识别这个值,然后在帮你自动减压后下载. 但是利用官方boto3包的download_fileobj()方法中,却无法指定对应的参数.下

codeAB 2019-10-23

Python中的可变对象与不可变对象、浅拷贝与深拷贝

Python中的对象分为可变与不可变，有必要了解一下，这会影响到python对象的赋值与拷贝。而拷贝也有深浅之别。简单说就是某个对象存放在内存中，这块内存中的值是不能改变的，变量指向这块内存，如果要改变变量的值，只能再开辟一块内存，放入新值，再让变量指向新

Laozizuiku 2019-10-20

S3 的一个命令行客户端

使用Golang 实现了一个S3的简易客户端。目前已发布1.0.0， 支持Windows和Linux，其他平台自行编译即可。Use "s3 <command> " for more information about a c

89961330 2019-09-07

AWS S3使用demo---java篇

AWS的api过于繁琐，入门者没必要去全文阅读api，我这里简单封装S3的api，方便大家上手使用s3。

yiranpiaoluo 2017-04-26

使用amazon的S3服务

今天帮同学解决一个问题,对方给了一个英文文档,intel那边只给了一些文章id,要把文章内容拿到在平板上进行展示.我看了下文档上提供了两种方案,第一种是通过AWS的方式.第二种直接通过URL把文章内容呈现出来.比较而言,第二种通过http请求,只需要生成t

yujs0000 2013-06-15

AWS存储服务特点与典型应用场景

存储是应用架构设计中非常关键的一环。不同的应用，对于存储的并发访问量、响应时间、数据持久性保护、高可用、访问接口以及成本等都有不同的需求。他还结合客户案例说明了这些服务的实际应用场景。EC2其实就是一个虚拟主机，大家是有完全的控制权限的。

xudawenfighting 2014-12-01

数学之美系列三：隐含马尔可夫模型在语言处理中的应用

复杂的语音识别问题通过隐含马尔可夫模型能非常简单地被表述、解决，让我不由由衷地感叹数学模型之妙。自然语言是人类交流信息的工具。以下该图就表示了一个典型的通信系统：。其中s1，s2，s3...表示信息源发出的信号。在利用隐含马尔可夫模型解决语言处理问题前，先

seasongirl 2012-03-08

AWS S3 静态对象存储

静态对象存储的意思就是，和FTP差不多，只能上传、下载、删除，不能修改、更新，更不能存动态数据库一类的东西。虽然还是有很多差别，但是S3和CDN差不多。为什么不选择可以当硬盘一样随便操作的EBS而是用静态对象存储的S3呢？便宜作为Archive存档还是非常

zhaoyu0 2019-06-30

AWS CLI 命令行直接操作AWS控制台

$ pip3 install awscli --upgrade --user建议在Virtualenv下安装，因为系统级别安装经过测试经常无法安装上。如同git和docker，awscli分为入口命令和功能命令，入口命令为aws，功能命令各自对应一项aws

Aws万岁爷 2019-06-30

AWS实战 - 对S3数据加密方式的总结

名词解释KMS: AWS Key Management Service，AWS的密钥托管服务；数据加密密钥: 用于对S3对象进行加密的密钥；主密钥: 也叫客户主密钥或CMK，根据不同的加密方式，主密钥有可能直接用于加密S3对象，也有可能用于加密数据加密密钥

XuYongshi0 2019-06-29

玩转全志/索智S3 Zeta Linux开发（1）框架描述及基础操作

在SDK设计时，Application、Core、Kernel彼此之间力求做到低耦合。在实际的项目研发当中，开发者/团队可以依靠数据字典/API聚焦于任一层次或任一模块，不相互影响。zmake all2.2 烧录固件我们把zmake all或zmake i

luoshuitianyi 2019-06-28

Consul 搭建集群

Consul 搭建集群Consul 是一套开源的分布式服务发现和配置管理系统，由 HashiCorp 公司用 Go 语言开发。它具有很多优点。快速入门环境准备首先需要准备好三台虚拟机,由三台虚机组成一个consul集群。==> s1: Matchin

wxtyixiao 2019-06-27
心中永远的正能量

心中永远的正能量

W3CSchool教程
HTML 教程
CSS 教程
Bootstrap 教程
Javascript 教程
jQuery 教程
后端教程
C 教程
Java 教程
PHP 教程
Python 教程
Go 教程
移动开发
Android 教程
Swift 教程
Kotlin 教程
jQuery Mobile 教程
ionic 教程
关于我们
新闻动态
联系方式
招聘英才
安科实验室
帮助与反馈

安科网(Ancii)，中国第一极客网

Copyright © 2013 - 2019 Ancii.com

京ICP备18063983号 京公网安备11010802014868号