`
bupt04406
  • 浏览: 347737 次
  • 性别: Icon_minigender_1
  • 来自: 杭州
社区版块
存档分类
最新评论

HBase如何从Hadoop读取数据,DFSInputStream

 
阅读更多

HDFS Client的读取流是从DFSInputStream来获得的,外层做了不少包装。

从DFSInputStream读取数据有两种方式:

(1)seek(long targetPos) + read(byte buf[], int off, int len)

(2)read(long position, byte[] buffer, int offset, int length)

 

第一种适合顺序读取,比如hbase里面的scan请求或者hbase的compact读取。读取的数据量一般都比较大,所以打开预读可以减少磁盘的iops,预读可以见hdfs相关的jira

Add HDFS support for fadvise readahead and drop-behind

https://issues.apache.org/jira/browse/HDFS-2465

Enable fadvise readahead by default 

https://issues.apache.org/jira/browse/HDFS-3697

 

 

第二种是随机读取,适合读取少量的数据,比如hbase里面的get请求

 

这两种读取还有一些不同的地方:

同一个DFSInputStream可以有多个应用程序在使用,但是需要注意下面:

(1)同一个时候只能有一个应用程序在执行seek(long targetPos) + read(byte buf[], int off, int len) 

(2)在一个应用程序执行seek(long targetPos) + read(byte buf[], int off, int len) 时,其他应用程序使用这个DFSInputStream执行read(long position, byte[] buffer, int offset, int length),并且同一时间可以有多个应用程序使用这个DFSInputStream执行read(long position, byte[] buffer, int offset, int length)

 

TestCase可以见hdfs的org.apache.hadoop.hdfs.TestPread

 

我们看下hbase是如何来使用的:

pread为true代表随机读取,当请求是get请求时会设置pread为true

0.90.x版本:

 

 

 

BoundedRangeFileInputStream
  @Override
  public int read(byte[] b, int off, int len) throws IOException {
    if ((off | len | (off + len) | (b.length - (off + len))) < 0) {
      throw new IndexOutOfBoundsException();
    }

    int n = (int) Math.min(Integer.MAX_VALUE, Math.min(len, (end - pos)));
    if (n == 0) return -1;
    int ret = 0;
    if (this.pread) { // 随机读取,第二种方式,可以任意多应用程序同时调用
      ret = in.read(pos, b, off, n);
    } else {
      synchronized (in) { //顺序读取,第一种方式,同一个时候只能有一个应用程序调用,所以这个地方加了锁
        in.seek(pos);
        ret = in.read(b, off, n);
      }
    }
    if (ret < 0) {
      end = pos;
      return -1;
    }
    pos += ret;
    return ret;
  }

 

 

 

 

0.94.x版本:

 

 

HFileBlock

    protected int readAtOffset(FSDataInputStream istream,
        byte[] dest, int destOffset, int size,
        boolean peekIntoNextBlock, long fileOffset, boolean pread)
        throws IOException {
      if (peekIntoNextBlock &&
          destOffset + size + hdrSize > dest.length) {
        // We are asked to read the next block's header as well, but there is
        // not enough room in the array.
        throw new IOException("Attempted to read " + size + " bytes and " +
            hdrSize + " bytes of next header into a " + dest.length +
            "-byte array at offset " + destOffset);
      }

      if (pread) {
        // Positional read. Better for random reads.
        int extraSize = peekIntoNextBlock ? hdrSize : 0;

        int ret = istream.read(fileOffset, dest, destOffset, size + extraSize);  // 随机读取,第二种方式,可以任意多应用程序同时调用
        if (ret < size) {
          throw new IOException("Positional read of " + size + " bytes " +
              "failed at offset " + fileOffset + " (returned " + ret + ")");
        }

        if (ret == size || ret < size + extraSize) {
          // Could not read the next block's header, or did not try.
          return -1;
        }
      } else {
        // Seek + read. Better for scanning.
        synchronized (istream) { //顺序读取,第一种方式,同一个时候只能有一个应用程序调用,所以这个地方加了锁
          istream.seek(fileOffset);

          long realOffset = istream.getPos();
          if (realOffset != fileOffset) {
            throw new IOException("Tried to seek to " + fileOffset + " to "
                + "read " + size + " bytes, but pos=" + realOffset
                + " after seek");
          }

          if (!peekIntoNextBlock) {
            IOUtils.readFully(istream, dest, destOffset, size);
            return -1;
          }

          // Try to read the next block header.
          if (!readWithExtra(istream, dest, destOffset, size, hdrSize))
            return -1;
        }
      }

      assert peekIntoNextBlock;
      return Bytes.toInt(dest, destOffset + size + BlockType.MAGIC_LENGTH) +
          hdrSize;
    }

 

 

为什么随机读取的时候可以任意多个应用程序使用呢:

 

(1)seek(long targetPos) + read(byte buf[], int off, int len) 

第一种,seek很简单改变下pos,read(byte buf[], int off, int len) 里面:重要的是blockSeekTo(long target) ,blockSeekTo第一步做的操作就是看下当前blockReader是否为null,如果不为null那么就close这个BlockReader,

然后new一个BlockReader,向DataNode发送的请求里面包含当前开始读取的位置,以及长度(blk.getNumBytes() - offsetIntoBlock来获得,也就是说长度是当前块剩余可读的数据量), read(byte buf[], int off, int len) 可以多次调用,直到数据已经读完。在这里我们可以看到new 了一个BlockReader后,没有及时关闭,BlockReader关闭的时机是下次调用seek+read时发现上次new 的BlockReader还在,那么就关闭。 所以如果一个应用程序在调用一个DFSInputStream流的seek+read时,如果另外一个应用执行seek+read的话,后面这个应用程序会把前面那个应用程序的BlockReader关闭,导致前面一个应用程序读取不到数据了,所以同一时间只能有一个应用程序调用一个DFSInputStream流的seek+read,除非使用另外一个DFSInputStream流的seek+read

 

(2)read(long position, byte[] buffer, int offset, int length)

第二种,读取时每次都会new一个BlockReader,向DataNode发送的请求里面包含了读取开始的位置(position),以及长度length,读取完后就会close这个BlockReader

 

 

 

pread Use positional read instead of seek+read (positional is better doing random reads whereas seek+read is better scanning).

 

 https://issues.apache.org/jira/browse/HBASE-2180

 /proc/sys/net/ipv4/tcp_tw_recycle

 

 

 

Allow multiple readers per storefile

https://issues.apache.org/jira/browse/HBASE-7347

 

Non-pread DFSInputStreams should be associated with scanners, not HFile.Readers

https://issues.apache.org/jira/browse/HBASE-5979

 

HFileBlock.readAtOffset does not work well with multiple threads

https://issues.apache.org/jira/browse/HBASE-7336

 

Consider double-checked locking for block cache lock

https://issues.apache.org/jira/browse/HBASE-5898

 

http://markmail.org/thread/ulki4uccwb43ahdj


[89-fb] Using pread for non-compaction read request

https://issues.apache.org/jira/browse/HBASE-7266

 

 

 

分享到:
评论

相关推荐

    hbase和hadoop数据块损坏处理

    HBase 和 Hadoop 数据块损坏处理 HBase 和 Hadoop 数据块损坏是非常常见的问题,可能会导致数据丢失、集群崩溃等严重后果。因此,了解如何处理 HBase 和 Hadoop 数据块损坏是非常重要的。本文将介绍 HBase 和 ...

    hbase-hadoop2-compat-1.2.12-API文档-中文版.zip

    赠送jar包:hbase-hadoop2-compat-1.2.12.jar; 赠送原API文档:hbase-hadoop2-compat-1.2.12-javadoc.jar; 赠送源代码:hbase-hadoop2-compat-1.2.12-sources.jar; 赠送Maven依赖信息文件:hbase-hadoop2-compat-...

    hbase-hadoop-compat-1.1.3-API文档-中文版.zip

    赠送jar包:hbase-hadoop-compat-1.1.3.jar; 赠送原API文档:hbase-hadoop-compat-1.1.3-javadoc.jar; 赠送源代码:hbase-hadoop-compat-1.1.3-sources.jar; 赠送Maven依赖信息文件:hbase-hadoop-compat-1.1.3....

    hbase-hadoop2-compat-1.1.3-API文档-中文版.zip

    赠送jar包:hbase-hadoop2-compat-1.1.3.jar; 赠送原API文档:hbase-hadoop2-compat-1.1.3-javadoc.jar; 赠送源代码:hbase-hadoop2-compat-1.1.3-sources.jar; 赠送Maven依赖信息文件:hbase-hadoop2-compat-...

    hbase-hadoop-compat-1.1.3-API文档-中英对照版.zip

    赠送jar包:hbase-hadoop-compat-1.1.3.jar; 赠送原API文档:hbase-hadoop-compat-1.1.3-javadoc.jar; 赠送源代码:hbase-hadoop-compat-1.1.3-sources.jar; 赠送Maven依赖信息文件:hbase-hadoop-compat-1.1.3....

    Hbase和Hadoop JMX监控实战

    在大数据处理领域,HBase和Hadoop是两个关键的组件,它们在分布式存储和处理大量数据方面发挥着重要作用。JMX(Java Management Extensions)是一种Java平台标准,用于管理和监控应用程序。在本实战中,我们将深入...

    hbase-hadoop2-compat-1.1.3-API文档-中英对照版.zip

    赠送jar包:hbase-hadoop2-compat-1.1.3.jar; 赠送原API文档:hbase-hadoop2-compat-1.1.3-javadoc.jar; 赠送源代码:hbase-hadoop2-compat-1.1.3-sources.jar; 赠送Maven依赖信息文件:hbase-hadoop2-compat-...

    hbase-hadoop-compat-1.2.12-API文档-中文版.zip

    赠送jar包:hbase-hadoop-compat-1.2.12.jar; 赠送原API文档:hbase-hadoop-compat-1.2.12-javadoc.jar; 赠送源代码:hbase-hadoop-compat-1.2.12-sources.jar; 赠送Maven依赖信息文件:hbase-hadoop-compat-...

    hbase-hadoop-compat-1.2.12-API文档-中英对照版.zip

    赠送jar包:hbase-hadoop-compat-1.2.12.jar; 赠送原API文档:hbase-hadoop-compat-1.2.12-javadoc.jar; 赠送源代码:hbase-hadoop-compat-1.2.12-sources.jar; 赠送Maven依赖信息文件:hbase-hadoop-compat-...

    hbase-hadoop1-compat-0.98.3-hadoop1.zip

    HBase基于谷歌的Bigtable设计,构建在Hadoop之上,提供实时读写访问海量结构化数据。它的主要特性包括强一致性、行式存储、稀疏性以及支持多种数据类型。Hadoop 1.x则包含了MapReduce和HDFS(Hadoop分布式文件系统)...

    hbase-hadoop-compat-1.4.3-API文档-中文版.zip

    赠送jar包:hbase-hadoop-compat-1.4.3.jar; 赠送原API文档:hbase-hadoop-compat-1.4.3-javadoc.jar; 赠送源代码:hbase-hadoop-compat-1.4.3-sources.jar; 赠送Maven依赖信息文件:hbase-hadoop-compat-1.4.3....

    hbase-hadoop2-compat-1.2.12-API文档-中英对照版.zip

    赠送jar包:hbase-hadoop2-compat-1.2.12.jar; 赠送原API文档:hbase-hadoop2-compat-1.2.12-javadoc.jar; 赠送源代码:hbase-hadoop2-compat-1.2.12-sources.jar; 赠送Maven依赖信息文件:hbase-hadoop2-compat-...

    hbase-hadoop2-compat-1.4.3-API文档-中文版.zip

    赠送jar包:hbase-hadoop2-compat-1.4.3.jar; 赠送原API文档:hbase-hadoop2-compat-1.4.3-javadoc.jar; 赠送源代码:hbase-hadoop2-compat-1.4.3-sources.jar; 赠送Maven依赖信息文件:hbase-hadoop2-compat-...

    hbase-hadoop+database系统入门书籍

    - **MapReduce进行数据处理**:HBase支持MapReduce作业直接读取和写入表数据,这使得对HBase中的数据进行复杂的分析成为可能。 - **Hadoop生态系统的其他组件**:除了HDFS和MapReduce之外,HBase还可以与其他Hadoop...

    Hadoop数据迁移--从Hadoop向HBase载入数据

    在Hadoop数据迁移过程中,从Hadoop向HBase载入数据是常见的操作。 数据迁移主要分为两个步骤: 第一步骤:将Hadoop中普通文本格式的数据转化为HFile文件。HFile是HBase中的存储格式文件,它可以被HBase识别和读取...

    java操作Hbase之从Hbase中读取数据写入hdfs中源码

    下面是一段简单的Java代码示例,演示如何从HBase中读取数据: ```java import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client....

    较全面的Hbase与hadoop分析报告.doc

    HBase 利用 Hadoop 的分布式存储能力,实现了对大规模数据的高效管理和查询,而 Hadoop 则通过 HDFS 提供了可靠的数据存储和计算平台。理解这两个组件的工作原理和相互关系对于优化大数据环境中的数据管理和分析至关...

    hbase&hadoop初探

    标题中的“hbase&hadoop初探”表明我们将要探讨的是大数据处理领域中的两个关键组件:HBase和Hadoop。这两个技术是Apache软件基金会的重要项目,对于理解和掌握大数据存储与处理有着至关重要的作用。 首先,Hadoop...

    hadoop-2.7.2-hbase-jar.tar.gz

    HBase的运行依赖于Hadoop的HDFS作为底层存储系统,MapReduce用于批量处理数据。"hadoop-2.7.2-hbase-jar"文件中的JAR包,包含HBase运行所需的类库和API,使得开发者可以在Hadoop集群上开发和运行HBase应用。 集成...

    hbase-0.90.5.tar.gz与hadoop0.20.2版本匹配

    确保HBase和Hadoop版本兼容是成功部署的关键,因为不兼容的版本可能会导致各种问题,如数据丢失、性能下降或服务不可用。在实际生产环境中,应该根据项目需求选择合适的HBase和Hadoop版本,以保证系统的稳定性和高效...

Global site tag (gtag.js) - Google Analytics