当前位置:首页 > 开发 > 系统架构 > 架构 > 正文

hbase increment代码

发表于: 2015-06-06   作者:blackproof   来源:转载   浏览次数:
摘要: hbase increase       increase代码 1.将数据封装为increment对象 2.从increment对象中封装get 3.封装新kv 4.对HRegion下的Storm做upsert或add操作 5.查看是否需要flush并添加队列 6.返回kvs   HRegion代码,如下   /**

hbase increase    

 

increase代码

1.将数据封装为increment对象

2.从increment对象中封装get

3.封装新kv

4.对HRegion下的Storm做upsert或add操作

5.查看是否需要flush并添加队列

6.返回kvs

 

HRegion代码,如下

 

  /**
   * Perform one or more increment operations on a row.
   * @param increment
   * @return new keyvalues after increment
   * @throws IOException
   */
  public Result increment(Increment increment, long nonceGroup, long nonce)
  throws IOException {
    byte [] row = increment.getRow();
    checkRow(row, "increment");
    TimeRange tr = increment.getTimeRange();
    boolean flush = false;
    Durability durability = getEffectiveDurability(increment.getDurability());
    boolean writeToWAL = durability != Durability.SKIP_WAL;
    WALEdit walEdits = null;
    List<Cell> allKVs = new ArrayList<Cell>(increment.size());
    Map<Store, List<Cell>> tempMemstore = new HashMap<Store, List<Cell>>();

    long size = 0;
    long txid = 0;

    checkReadOnly();
    checkResources();
    // Lock row
    startRegionOperation(Operation.INCREMENT);
    this.writeRequestsCount.increment();
    WriteEntry w = null;
    try {
      RowLock rowLock = getRowLock(row);
      try {
        lock(this.updatesLock.readLock());
        // wait for all prior MVCC transactions to finish - while we hold the row lock
        // (so that we are guaranteed to see the latest state)
        mvcc.completeMemstoreInsert(mvcc.beginMemstoreInsert());
        // now start my own transaction
        w = mvcc.beginMemstoreInsert();
        try {
          long now = EnvironmentEdgeManager.currentTimeMillis();
          // Process each family
          for (Map.Entry<byte [], List<Cell>> family:
              increment.getFamilyCellMap().entrySet()) {

            Store store = stores.get(family.getKey());
            List<Cell> kvs = new ArrayList<Cell>(family.getValue().size());

            // Sort the cells so that they match the order that they
            // appear in the Get results. Otherwise, we won't be able to
            // find the existing values if the cells are not specified
            // in order by the client since cells are in an array list.
            Collections.sort(family.getValue(), store.getComparator());
            // Get previous values for all columns in this family
            // 从increment封装get请求
            Get get = new Get(row);
            for (Cell cell: family.getValue()) {
              KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
              get.addColumn(family.getKey(), kv.getQualifier());
            }
            get.setTimeRange(tr.getMin(), tr.getMax());
            List<Cell> results = get(get, false);//获得此increase的row

            // Iterate the input columns and update existing values if they were
            // found, otherwise add new column initialized to the increment amount
            int idx = 0;
            for (Cell kv: family.getValue()) {
              long amount = Bytes.toLong(CellUtil.cloneValue(kv));//获得当前value值
              boolean noWriteBack = (amount == 0);

              Cell c = null;
              if (idx < results.size() && CellUtil.matchingQualifier(results.get(idx), kv)) {
                c = results.get(idx);
                if(c.getValueLength() == Bytes.SIZEOF_LONG) {
                  amount += Bytes.toLong(c.getValueArray(), c.getValueOffset(), Bytes.SIZEOF_LONG);
                } else {
                  // throw DoNotRetryIOException instead of IllegalArgumentException
                  throw new org.apache.hadoop.hbase.DoNotRetryIOException(
                      "Attempted to increment field that isn't 64 bits wide");
                }
                idx++;
              }

              // Append new incremented KeyValue to list
              byte[] q = CellUtil.cloneQualifier(kv);
              byte[] val = Bytes.toBytes(amount);
              int oldCellTagsLen = (c == null) ? 0 : c.getTagsLength();
              int incCellTagsLen = kv.getTagsLength();
              KeyValue newKV = new KeyValue(row.length, family.getKey().length, q.length, now,
                  KeyValue.Type.Put, val.length, oldCellTagsLen + incCellTagsLen);
              System.arraycopy(row, 0, newKV.getBuffer(), newKV.getRowOffset(), row.length);
              System.arraycopy(family.getKey(), 0, newKV.getBuffer(), newKV.getFamilyOffset(),
                  family.getKey().length);
              System.arraycopy(q, 0, newKV.getBuffer(), newKV.getQualifierOffset(), q.length);
              // copy in the value
              System.arraycopy(val, 0, newKV.getBuffer(), newKV.getValueOffset(), val.length);
              // copy tags
              if (oldCellTagsLen > 0) {
                System.arraycopy(c.getTagsArray(), c.getTagsOffset(), newKV.getBuffer(),
                    newKV.getTagsOffset(), oldCellTagsLen);
              }
              if (incCellTagsLen > 0) {
                System.arraycopy(kv.getTagsArray(), kv.getTagsOffset(), newKV.getBuffer(),
                    newKV.getTagsOffset() + oldCellTagsLen, incCellTagsLen);
              }
              newKV.setMvccVersion(w.getWriteNumber());
              // Give coprocessors a chance to update the new cell
              if (coprocessorHost != null) {
                newKV = KeyValueUtil.ensureKeyValue(coprocessorHost.postMutationBeforeWAL(
                    RegionObserver.MutationType.INCREMENT, increment, c, (Cell) newKV));
              }
              allKVs.add(newKV);

              if (!noWriteBack) {
                kvs.add(newKV);

                // Prepare WAL updates
                if (writeToWAL) {
                  if (walEdits == null) {
                    walEdits = new WALEdit();
                  }
                  walEdits.add(newKV);
                }
              }
            }

            //store the kvs to the temporary memstore before writing HLog
            if (!kvs.isEmpty()) {
              tempMemstore.put(store, kvs);
            }
          }

          // Actually write to WAL now
          if (walEdits != null && !walEdits.isEmpty()) {
            if (writeToWAL) {
              // Using default cluster id, as this can only happen in the orginating
              // cluster. A slave cluster receives the final value (not the delta)
              // as a Put.
              txid = this.log.appendNoSync(this.getRegionInfo(),
                  this.htableDescriptor.getTableName(), walEdits, new ArrayList<UUID>(),
                  EnvironmentEdgeManager.currentTimeMillis(), this.htableDescriptor, this.sequenceId,
                  true, nonceGroup, nonce);
            } else {
              recordMutationWithoutWal(increment.getFamilyCellMap());
            }
          }
          //Actually write to Memstore now
          if (!tempMemstore.isEmpty()) {//更新hbase kv
            for (Map.Entry<Store, List<Cell>> entry : tempMemstore.entrySet()) {
              Store store = entry.getKey();
              if (store.getFamily().getMaxVersions() == 1) {
                // upsert if VERSIONS for this CF == 1
                size += store.upsert(entry.getValue(), getSmallestReadPoint());
              } else {
                // otherwise keep older versions around
                for (Cell cell : entry.getValue()) {
                  KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
                  size += store.add(kv);
                }
              }
            }
            size = this.addAndGetGlobalMemstoreSize(size);
            flush = isFlushSize(size);
          }
        } finally {
          this.updatesLock.readLock().unlock();
        }
      } finally {
        rowLock.release();
      }
      if (writeToWAL && (walEdits != null) && !walEdits.isEmpty()) {
        // sync the transaction log outside the rowlock
        syncOrDefer(txid, durability);
      }
    } finally {
      if (w != null) {
        mvcc.completeMemstoreInsert(w);
      }
      closeRegionOperation(Operation.INCREMENT);
      if (this.metricsRegion != null) {
        this.metricsRegion.updateIncrement();
      }
    }

    if (flush) {
      // Request a cache flush.  Do it outside update lock.
      requestFlush();
    }

    return Result.create(allKVs);
  }

 

在一些情况下,如increment压力过大时,会出现下列错误,startNonceOperation方法:

 

 regionserver.ServerNonceManager: Conflict detected by nonce

 

一个mutation里边有多个相同的nonce的操作,如increment,这样会产生此日志,影响相应速度

 

 

 

hbase increment代码

  • 0

    开心

    开心

  • 0

    板砖

    板砖

  • 0

    感动

    感动

  • 0

    有用

    有用

  • 0

    疑问

    疑问

  • 0

    难过

    难过

  • 0

    无聊

    无聊

  • 0

    震惊

    震惊

编辑推荐
在 HBase: The Definitive Guide 中, Lars George 介绍了 HBase 的一个新特性 Counter Increment,
1、引入源代码 下载hbase的源码,用mvn eclipse:eclipse来使工程可导入eclipse,其中需要安装protoc
1、引入源代码 下载hbase的源码,用mvn eclipse:eclipse来使工程可导入eclipse,其中需要安装protoc
一、脚本 start-hbase.sh,hbase-daemon.sh,hbase-daemons.sh,zookeepers.sh,regionservers.sh,hbase
随着HBase的大规模应用,HBase的容灾显得特别的重要。 本文主要从代码层面分析HBase replication (基
原创文章,请各位多多指导,有错误希望各位能及时告诉我,感激不尽~ (1) 前面我们说了,locateReg
原创文章,请各位多多指导,有错误希望各位能及时告诉我,感激不尽~ (1) 前面我们说了,locateReg
7.存储模型 regionserver负责region内数据的存储及处理客户端的读写请求: 写请求:当regionserver接
Java,操作Redis,标配Jedis,如果加上Spring,自然选用Spring Data Redis(简称“sdr”)。 最近在
问题描述: SSH框架开发中,Hibernate查询SQLServer中的数据没有问题,但是插入数据会报错。 报错:
版权所有 IT知识库 CopyRight © 2009-2015 IT知识库 IT610.com , All Rights Reserved. 京ICP备09083238号