org.apache.hadoop.hbase.Cell#getValue ( )源码实例Demo

/**
	 * Asynchronous HBase scan read as RAW qualifier
	 *
	 * @param scan
	 * @param listener
	 * @throws Exception
	 */
	protected InternalReadReport asyncStreamRead(EntityDefinition ed, Scan scan, QualifierCreationListener listener) throws IOException {
//		_init();
		long counter = 0;
		long startTimestamp = 0;
		long stopTimestamp = 0;
		InternalScanner scanner = this.getCurrentRegion().getScanner(scan);
		List<Cell> results = new ArrayList<Cell>();
		try{
			boolean hasMoreRows;//false by default
			do{
				hasMoreRows = scanner.next(results);
				Map<String, byte[]> kvMap = new HashMap<String, byte[]>();
				if(!results.isEmpty()){
					counter ++;
					byte[] row = results.get(0).getRow();
//					if(ed.isTimeSeries()){
					long timestamp = RowkeyBuilder.getTimestamp(row,ed);
					// Min
					if(startTimestamp == 0 || startTimestamp > timestamp ){
						startTimestamp = timestamp;
					}
					// Max
					if(stopTimestamp == 0 || stopTimestamp < timestamp ){
						stopTimestamp = timestamp;
					}
//					}
					
					for(Cell kv:results){
						String qualifierName = Bytes.toString(kv.getQualifier());
						Qualifier qualifier = null;
						if(!ed.isTag(qualifierName)){
							qualifier = ed.getQualifierNameMap().get(qualifierName);
							if(qualifier == null){
								LOG.error("qualifier for field " + qualifierName + " not exist");
								throw new IOException(new NullPointerException("qualifier for field "+qualifierName+" is null"));
							}
							qualifierName = qualifier.getDisplayName();
						}
						if(kv.getValue()!=null) kvMap.put(qualifierName,kv.getValue());
					}
					
//					LOG.info("DEBUG: timestamp="+timestamp+", keys=["+StringUtils.join(kvMap.keySet(),",")+"]");

					if(!kvMap.isEmpty()) listener.qualifierCreated(kvMap);
					results.clear();
				}else{
					if(LOG.isDebugEnabled()) LOG.warn("Empty batch of KeyValue");
				}
			} while(hasMoreRows);
		} catch(IOException ex){
			LOG.error(ex.getMessage(),ex);
			throw ex;
		} finally {
            if(scanner != null) {
                scanner.close();
            }
		}

		return new InternalReadReport(counter,startTimestamp,stopTimestamp);
	}
 

/**
 * Asynchronous HBase scan read as RAW qualifier.
 */
protected InternalReadReport asyncStreamRead(EntityDefinition ed, Scan scan, QualifierCreationListener listener) throws IOException {
    long counter = 0;
    long startTimestamp = 0;
    long stopTimestamp = 0;
    InternalScanner scanner = this.getCurrentRegion().getScanner(scan);
    List<Cell> results = new ArrayList<Cell>();
    try {
        boolean hasMoreRows;//false by default
        do {
            hasMoreRows = scanner.next(results);
            Map<String, byte[]> kvMap = new HashMap<>();
            if (!results.isEmpty()) {
                counter++;
                byte[] row = results.get(0).getRow();
                long timestamp = RowkeyBuilder.getTimestamp(row, ed);
                // Min
                if (startTimestamp == 0 || startTimestamp > timestamp) {
                    startTimestamp = timestamp;
                }
                // Max
                if (stopTimestamp == 0 || stopTimestamp < timestamp) {
                    stopTimestamp = timestamp;
                }

                for (Cell kv : results) {
                    String qualifierName = Bytes.toString(kv.getQualifier());
                    Qualifier qualifier = null;
                    if (!ed.isTag(qualifierName)) {
                        qualifier = ed.getQualifierNameMap().get(qualifierName);
                        if (qualifier == null) {
                            LOG.error("qualifier for field " + qualifierName + " not exist");
                            throw new IOException(new NullPointerException("qualifier for field " + qualifierName + " is null"));
                        }
                        qualifierName = qualifier.getDisplayName();
                    }
                    if (kv.getValue() != null) {
                        kvMap.put(qualifierName, kv.getValue());
                    }
                }

                if (!kvMap.isEmpty()) {
                    listener.qualifierCreated(kvMap);
                }
                results.clear();
            } else {
                if (LOG.isDebugEnabled()) {
                    LOG.warn("Empty batch of KeyValue");
                }
            }
        }
        while (hasMoreRows);
    } catch (IOException ex) {
        LOG.error(ex.getMessage(), ex);
        throw ex;
    } finally {
        if (scanner != null) {
            scanner.close();
        }
    }

    return new InternalReadReport(counter, startTimestamp, stopTimestamp);
}
 

private void addColumnToTable(List<Cell> results, PName colName, PName famName,
    Cell[] colKeyValues, List<PColumn> columns, boolean isSalted) {
    int i = 0;
    int j = 0;
    while (i < results.size() && j < COLUMN_KV_COLUMNS.size()) {
        Cell kv = results.get(i);
        Cell searchKv = COLUMN_KV_COLUMNS.get(j);
        int cmp =
                Bytes.compareTo(kv.getQualifierArray(), kv.getQualifierOffset(),
                    kv.getQualifierLength(), searchKv.getQualifierArray(),
                    searchKv.getQualifierOffset(), searchKv.getQualifierLength());
        if (cmp == 0) {
            colKeyValues[j++] = kv;
            i++;
        } else if (cmp > 0) {
            colKeyValues[j++] = null;
        } else {
            i++; // shouldn't happen - means unexpected KV in system table column row
        }
    }

    if (colKeyValues[DATA_TYPE_INDEX] == null || colKeyValues[NULLABLE_INDEX] == null
            || colKeyValues[ORDINAL_POSITION_INDEX] == null) {
        throw new IllegalStateException("Didn't find all required key values in '"
                + colName.getString() + "' column metadata row");
    }

    Cell columnSizeKv = colKeyValues[COLUMN_SIZE_INDEX];
    Integer maxLength =
            columnSizeKv == null ? null : PInteger.INSTANCE.getCodec().decodeInt(
                columnSizeKv.getValueArray(), columnSizeKv.getValueOffset(), SortOrder.getDefault());
    Cell decimalDigitKv = colKeyValues[DECIMAL_DIGITS_INDEX];
    Integer scale =
            decimalDigitKv == null ? null : PInteger.INSTANCE.getCodec().decodeInt(
                decimalDigitKv.getValueArray(), decimalDigitKv.getValueOffset(), SortOrder.getDefault());
    Cell ordinalPositionKv = colKeyValues[ORDINAL_POSITION_INDEX];
    int position =
        PInteger.INSTANCE.getCodec().decodeInt(ordinalPositionKv.getValueArray(),
                ordinalPositionKv.getValueOffset(), SortOrder.getDefault()) + (isSalted ? 1 : 0);
    Cell nullableKv = colKeyValues[NULLABLE_INDEX];
    boolean isNullable =
        PInteger.INSTANCE.getCodec().decodeInt(nullableKv.getValueArray(),
                nullableKv.getValueOffset(), SortOrder.getDefault()) != ResultSetMetaData.columnNoNulls;
    Cell dataTypeKv = colKeyValues[DATA_TYPE_INDEX];
    PDataType dataType =
            PDataType.fromTypeId(PInteger.INSTANCE.getCodec().decodeInt(
              dataTypeKv.getValueArray(), dataTypeKv.getValueOffset(), SortOrder.getDefault()));
    if (maxLength == null && dataType == PBinary.INSTANCE) dataType = PVarbinary.INSTANCE;   // For
                                                                                           // backward
                                                                                           // compatibility.
    Cell sortOrderKv = colKeyValues[SORT_ORDER_INDEX];
    SortOrder sortOrder =
    		sortOrderKv == null ? SortOrder.getDefault() : SortOrder.fromSystemValue(PInteger.INSTANCE
                    .getCodec().decodeInt(sortOrderKv.getValueArray(),
                    		sortOrderKv.getValueOffset(), SortOrder.getDefault()));

    Cell arraySizeKv = colKeyValues[ARRAY_SIZE_INDEX];
    Integer arraySize = arraySizeKv == null ? null :
        PInteger.INSTANCE.getCodec().decodeInt(arraySizeKv.getValueArray(), arraySizeKv.getValueOffset(), SortOrder.getDefault());

    Cell viewConstantKv = colKeyValues[VIEW_CONSTANT_INDEX];
    byte[] viewConstant = viewConstantKv == null ? null : viewConstantKv.getValue();
    Cell isViewReferencedKv = colKeyValues[IS_VIEW_REFERENCED_INDEX];
    boolean isViewReferenced = isViewReferencedKv != null && Boolean.TRUE.equals(PBoolean.INSTANCE.toObject(isViewReferencedKv.getValueArray(), isViewReferencedKv.getValueOffset(), isViewReferencedKv.getValueLength()));
    Cell columnDefKv = colKeyValues[COLUMN_DEF_INDEX];
    String expressionStr = columnDefKv==null ? null : (String)PVarchar.INSTANCE.toObject(columnDefKv.getValueArray(), columnDefKv.getValueOffset(), columnDefKv.getValueLength());
    PColumn column = new PColumnImpl(colName, famName, dataType, maxLength, scale, isNullable, position-1, sortOrder, arraySize, viewConstant, isViewReferenced, expressionStr);
    columns.add(column);
}
 

@Nullable
@Override
public Map.Entry<byte[], byte[]> apply(@Nullable Cell cell) {
  return new AbstractMap.SimpleEntry<>(cell.getQualifier(), cell.getValue());
}