Java 类org.apache.hadoop.hbase.util.Bytes 实例源码

private void printResult(Result result) {
  long rowkey = Bytes.toInt(result.getRow());
  //    int a = Bytes.toInt(result.getValue(familyName, Bytes.toBytes("a")));
  //    int b = Bytes.toInt(result.getValue(familyName, Bytes.toBytes("b")));
  //    int c = Bytes.toInt(result.getValue(familyName, Bytes.toBytes("c")));
  //    int info = Bytes.toInt(result.getValue(familyName, Bytes.toBytes("info")));
  StringBuilder sb = new StringBuilder();
  sb.append("{");
  sb.append("rowkey=").append(rowkey).append(",");

  for (Cell cell : result.listCells()) {
    sb.append(Bytes.toString(cell.getQualifier())).append("=")
        .append(Bytes.toInt(cell.getValue())).append(",");
  }

  //    sb.append("a=").append(a).append(",");
  //    sb.append("b=").append(b).append(",");
  //    sb.append("c=").append(c).append(",");
  //    sb.append("info=").append(info).append(",");
  sb.append("}");
  System.out.println(sb.toString());
}

/**
 * Use for logging.
 * @param b Key portion of a KeyValue.
 * @param o Offset to start of key
 * @param l Length of key.
 * @return Key as a String.
 */
public static String keyToString(final byte [] b, final int o, final int l) {
  if (b == null) return "";
  int rowlength = Bytes.toShort(b, o);
  String row = Bytes.toStringBinary(b, o + Bytes.SIZEOF_SHORT, rowlength);
  int columnoffset = o + Bytes.SIZEOF_SHORT + 1 + rowlength;
  int familylength = b[columnoffset - 1];
  int columnlength = l - ((columnoffset - o) + TIMESTAMP_TYPE_SIZE);
  String family = familylength == 0? "":
    Bytes.toStringBinary(b, columnoffset, familylength);
  String qualifier = columnlength == 0? "":
    Bytes.toStringBinary(b, columnoffset + familylength,
    columnlength - familylength);
  long timestamp = Bytes.toLong(b, o + (l - TIMESTAMP_TYPE_SIZE));
  String timestampStr = humanReadableTimestamp(timestamp);
  byte type = b[o + l - 1];
  return row + "/" + family +
    (family != null && family.length() > 0? ":" :"") +
    qualifier + "/" + timestampStr + "/" + Type.codeToType(type);
}

/**
 * @throws IllegalArgumentException if fullName equals old root or old meta. Some code
 *  depends on this. The test is buried in the table creation to save on array comparison
 *  when we're creating a standard table object that will be in the cache.
 */
public static TableName valueOf(byte[] fullName) throws IllegalArgumentException{
  for (TableName tn : tableCache) {
    if (Arrays.equals(tn.getName(), fullName)) {
      return tn;
    }
  }

  int namespaceDelimIndex = com.google.common.primitives.Bytes.lastIndexOf(fullName,
      (byte) NAMESPACE_DELIM);

  if (namespaceDelimIndex < 0) {
    return createTableNameIfNecessary(
        ByteBuffer.wrap(NamespaceDescriptor.DEFAULT_NAMESPACE_NAME),
        ByteBuffer.wrap(fullName));
  } else {
    return createTableNameIfNecessary(
        ByteBuffer.wrap(fullName, 0, namespaceDelimIndex),
        ByteBuffer.wrap(fullName, namespaceDelimIndex + 1,
            fullName.length - (namespaceDelimIndex + 1)));
  }
}

/**
 * Verify that Result.compareResults(...) behaves correctly.
 */
public void testCompareResults() throws Exception {
  byte [] value1 = Bytes.toBytes("value1");
  byte [] qual = Bytes.toBytes("qual");

  KeyValue kv1 = new KeyValue(row, family, qual, value);
  KeyValue kv2 = new KeyValue(row, family, qual, value1);

  Result r1 = Result.create(new KeyValue[] {kv1});
  Result r2 = Result.create(new KeyValue[] {kv2});
  // no exception thrown
  Result.compareResults(r1, r1);
  try {
    // these are different (HBASE-4800)
    Result.compareResults(r1, r2);
    fail();
  } catch (Exception x) {
    assertTrue(x.getMessage().startsWith("This result was different:"));
  }
}

@Override
public void failedBulkLoad(final byte[] family, final String srcPath) throws IOException {
  if (!FSHDFSUtils.isSameHdfs(conf, srcFs, fs)) {
    // files are copied so no need to move them back
    return;
  }
  Path p = new Path(srcPath);
  Path stageP = new Path(stagingDir,
      new Path(Bytes.toString(family), p.getName()));
  LOG.debug("Moving " + stageP + " back to " + p);
  if(!fs.rename(stageP, p))
    throw new IOException("Failed to move HFile: " + stageP + " to " + p);

  // restore original permission
  if (origPermissions.containsKey(srcPath)) {
    fs.setPermission(p, origPermissions.get(srcPath));
  } else {
    LOG.warn("Can't find previous permission for path=" + srcPath);
  }
}

private static void checkExistence(HTable htable, byte[] row, byte[] family, byte[] quality)
    throws Exception {
  // verify that the Get returns the correct result
  Result r;
  Get get = new Get(row);
  get.addColumn(FAMILY, QUALIFIER);
  int nbTry = 0;
  do {
    assertTrue("Fail to get from " + htable.getName() + " after " + nbTry + " tries", nbTry < 50);
    nbTry++;
    Thread.sleep(100);
    r = htable.get(get);
  } while (r == null || r.getValue(FAMILY, QUALIFIER) == null);
  assertEquals("value", Bytes.toStringBinary(VALUE1),
    Bytes.toStringBinary(r.getValue(FAMILY, QUALIFIER)));
}

/**
 * Create a new reader from the split, and match the edits against the passed columns.
 */
private void testSplit(InputSplit split, byte[]... columns) throws Exception {
  final WALRecordReader reader = getReader();
  reader.initialize(split, MapReduceTestUtil.createDummyMapTaskAttemptContext(conf));

  for (byte[] column : columns) {
    assertTrue(reader.nextKeyValue());
    Cell cell = reader.getCurrentValue().getCells().get(0);
    if (!Bytes.equals(column, cell.getQualifier())) {
      assertTrue("expected [" + Bytes.toString(column) + "], actual ["
          + Bytes.toString(cell.getQualifier()) + "]", false);
    }
  }
  assertFalse(reader.nextKeyValue());
  reader.close();
}

static String stringifyKvs(Collection<Cell> kvs) {
  StringBuilder out = new StringBuilder();
  out.append("[");
  if (kvs != null) {
    for (Cell kv : kvs) {
      byte[] col = CellUtil.cloneQualifier(kv);
      byte[] val = CellUtil.cloneValue(kv);
      if (Bytes.equals(col, COUNTER)) {
        out.append(Bytes.toStringBinary(col) + ":" +
                   Bytes.toInt(val) + " ");
      } else {
        out.append(Bytes.toStringBinary(col) + ":" +
                   Bytes.toStringBinary(val) + " ");
      }
    }
  }
  out.append("]");
  return out.toString();
}

public static Table createTableAndWriteDataWithLabels(TableName tableName, String... labelExps)
    throws Exception {
  Table table = null;
  table = TEST_UTIL.createTable(tableName, fam);
  int i = 1;
  List<Put> puts = new ArrayList<Put>();
  for (String labelExp : labelExps) {
    Put put = new Put(Bytes.toBytes("row" + i));
    put.add(fam, qual, HConstants.LATEST_TIMESTAMP, value);
    put.setCellVisibility(new CellVisibility(labelExp));
    puts.add(put);
    table.put(put);
    i++;
  }
  // table.put(puts);
  return table;
}

public void testMultiVersionLoadValue() throws Exception {
  KeyValue [] kvs1 = genKVs(row, family, value, 1, 100);
  KeyValue [] kvs2 = genKVs(row, family, value, 200, 100);

  KeyValue [] kvs = new KeyValue[kvs1.length+kvs2.length];
  System.arraycopy(kvs1, 0, kvs, 0, kvs1.length);
  System.arraycopy(kvs2, 0, kvs, kvs1.length, kvs2.length);

  Arrays.sort(kvs, KeyValue.COMPARATOR);

  ByteBuffer loadValueBuffer = ByteBuffer.allocate(1024);

  Result r = Result.create(kvs);
  for (int i = 0; i < 100; ++i) {
    final byte[] qf = Bytes.toBytes(i);

    loadValueBuffer.clear();
    r.loadValue(family, qf, loadValueBuffer);
    loadValueBuffer.flip();
    assertEquals(ByteBuffer.wrap(Bytes.add(value, Bytes.toBytes(i))), loadValueBuffer);
    assertEquals(ByteBuffer.wrap(Bytes.add(value, Bytes.toBytes(i))),
        r.getValueAsByteBuffer(family, qf));
  }
}

@Test
public void testScanRowWithStrong() throws IOException {
    clean();
    String row = rowPrefix + 0;
    putRow(row + 0, 1000);
    putRow(row + 3, 1000);

    Scan scan = new Scan();
    scan.setConsistency(Consistency.STRONG);
    scan.setStartRow(Bytes.toBytes(row + 0));
    scan.setStopRow(Bytes.toBytes(row + 3));
    ResultScanner scanResult = table.getScanner(scan);

    Result result = scanResult.next();
    assertTrue(result != null);
}

public void verifyNumericRows(Table table, final byte[] f, int startRow, int endRow,
    int replicaId)
    throws IOException {
  for (int i = startRow; i < endRow; i++) {
    String failMsg = "Failed verification of row :" + i;
    byte[] data = Bytes.toBytes(String.valueOf(i));
    Get get = new Get(data);
    get.setReplicaId(replicaId);
    get.setConsistency(Consistency.TIMELINE);
    Result result = table.get(get);
    assertTrue(failMsg, result.containsColumn(f, null));
    assertEquals(failMsg, result.getColumnCells(f, null).size(), 1);
    Cell cell = result.getColumnLatestCell(f, null);
    assertTrue(failMsg,
      Bytes.equals(data, 0, data.length, cell.getValueArray(), cell.getValueOffset(),
        cell.getValueLength()));
  }
}

@Test
public void testBasic() throws Exception {
  OColumnValue[] cvs = genCVs(value, 1, 100);

  Arrays.sort(cvs, OColumnValue.KEY_COMPARATOR);

  OResult r = new OResult(cvs);

  for (int i = 0; i < 100; ++i) {
    final byte[] qf = Bytes.toBytes(i);

    List<OColumnValue> ks = r.getColumn(qf);
    assertEquals(1, ks.size());
    assertByteEquals(qf, ks.get(0).getQualifier());

    assertEquals(ks.get(0), r.getColumnLatest(qf));
    assertByteEquals(Bytes.add(value, Bytes.toBytes(i)), r.getValue(qf));
    assertTrue(r.containsColumn(qf));
  }
}

/**
 * Test KeyValues with negative timestamp.
 * 
 * @throws IOException
 *           On test failure.
 */
@Test
public void testNegativeTimestamps() throws IOException {
  List<KeyValue> kvList = new ArrayList<KeyValue>();
  byte[] row = new byte[0];
  byte[] family = new byte[0];
  byte[] qualifier = new byte[0];
  byte[] value = new byte[0];
  if (includesTags) {
    byte[] metaValue1 = Bytes.toBytes("metaValue1");
    byte[] metaValue2 = Bytes.toBytes("metaValue2");
    kvList.add(new KeyValue(row, family, qualifier, 0l, value, new Tag[] { new Tag((byte) 1,
        metaValue1) }));
    kvList.add(new KeyValue(row, family, qualifier, 0l, value, new Tag[] { new Tag((byte) 1,
        metaValue2) }));
  } else {
    kvList.add(new KeyValue(row, family, qualifier, -1l, Type.Put, value));
    kvList.add(new KeyValue(row, family, qualifier, -2l, Type.Put, value));
  }
  testEncodersOnDataset(kvList, includesMemstoreTS, includesTags);
}

@Test(expected=UnsupportedOperationException.class)
public void testScanRowWithCommitted() throws IOException {
    clean();
    String row = rowPrefix + 0;
    putRow(row + 0, 1000);
    putRow(row + 3, 1000);

    Scan scan = new Scan();
    scan.setIsolationLevel(IsolationLevel.READ_COMMITTED);
    scan.setStartRow(Bytes.toBytes(row + 0));
    scan.setStopRow(Bytes.toBytes(row + 3));
    ResultScanner scanResult = table.getScanner(scan);

    Result result = scanResult.next();
    assertTrue(result != null);
}

/**
 * Compares left to right assuming that left,loffset,llength and right,roffset,rlength are
 * full KVs laid out in a flat byte[]s.
 * @param left
 * @param loffset
 * @param llength
 * @param right
 * @param roffset
 * @param rlength
 * @return  0 if equal, <0 if left smaller, >0 if right smaller
 */
public int compareFlatKey(byte[] left, int loffset, int llength,
    byte[] right, int roffset, int rlength) {
  // Compare row
  short lrowlength = Bytes.toShort(left, loffset);
  short rrowlength = Bytes.toShort(right, roffset);
  int compare = compareRows(left, loffset + Bytes.SIZEOF_SHORT,
      lrowlength, right, roffset + Bytes.SIZEOF_SHORT, rrowlength);
  if (compare != 0) {
    return compare;
  }

  // Compare the rest of the two KVs without making any assumptions about
  // the common prefix. This function will not compare rows anyway, so we
  // don't need to tell it that the common prefix includes the row.
  return compareWithoutRow(0, left, loffset, llength, right, roffset,
      rlength, rrowlength);
}

@Test
public void testMutationWithOneDelete() throws IOException {
    clean();

    Put put = new Put(Bytes.toBytes("pk0"));
    put.addColumn(Bytes.toBytes(familyName), Bytes.toBytes("col_1"), Bytes.toBytes("col_1_var"));
    put.addColumn(Bytes.toBytes(familyName), Bytes.toBytes("col_2"), Bytes.toBytes("col_2_var"));
    table.put(put);


    Delete delete = new Delete(Bytes.toBytes("pk0"));
    delete.addColumns(Bytes.toBytes(familyName), Bytes.toBytes("col_2"));

    RowMutations mutaions = new RowMutations(Bytes.toBytes("pk0"));
    mutaions.add(delete);

    table.mutateRow(mutaions);

    Get get = new Get(Bytes.toBytes("pk0"));
    Result result = table.get(get);
    String value = Bytes.toString(result.getValue(Bytes.toBytes(familyName), Bytes.toBytes("col_1")));
    assertEquals("col_1_var", value);
    byte[] col2 = result.getValue(Bytes.toBytes(familyName), Bytes.toBytes("col_2"));
    assertTrue(col2 == null);
}

@Test
public void testJira6912() throws Exception {
  TableName TABLE = TableName.valueOf("testJira6912");
  Table foo = TEST_UTIL.createTable(TABLE, new byte[][] {FAMILY}, 10);

  List<Put> puts = new ArrayList<Put>();
  for (int i=0;i !=100; i++){
    Put put = new Put(Bytes.toBytes(i));
    put.add(FAMILY, FAMILY, Bytes.toBytes(i));
    puts.add(put);
  }
  foo.put(puts);
  // If i comment this out it works
  TEST_UTIL.flush();

  Scan scan = new Scan();
  scan.setStartRow(Bytes.toBytes(1));
  scan.setStopRow(Bytes.toBytes(3));
  scan.addColumn(FAMILY, FAMILY);
  scan.setFilter(new RowFilter(CompareFilter.CompareOp.NOT_EQUAL, new BinaryComparator(Bytes.toBytes(1))));

  ResultScanner scanner = foo.getScanner(scan);
  Result[] bar = scanner.next(100);
  assertEquals(1, bar.length);
}

protected String resultToString(Result result) {
  StringBuilder sb = new StringBuilder();
  sb.append("{").append(keyToString(result.getRow())).append(":");
  for (Cell cell : result.listCells()) {
    byte[] f = CellUtil.cloneFamily(cell);
    byte[] q = CellUtil.cloneQualifier(cell);
    RangeDescription range = rangeMap.get(Bytes.add(f, q));
    sb.append("[").append(Bytes.toString(f)).append(":").append(Bytes.toString(q)).append("->");
    if (notPrintingSet.contains(q)) sb.append("skipped random value");
    else sb.append(DataType.byteToString(range.dataType, CellUtil.cloneValue(cell)));
    sb.append("]");
  }
  sb.append("}");
  return sb.toString();
}

private static void loadData(final Table ht, final byte[][] families,
    final int rows, final int flushes) throws IOException {
  List<Put> puts = new ArrayList<Put>(rows);
  byte[] qualifier = Bytes.toBytes("val");
  for (int i = 0; i < flushes; i++) {
    for (int k = 0; k < rows; k++) {
      byte[] row = Bytes.toBytes(random.nextLong());
      Put p = new Put(row);
      for (int j = 0; j < families.length; ++j) {
        p.add(families[ j ], qualifier, row);
      }
      puts.add(p);
    }
    ht.put(puts);
    TEST_UTIL.flush();
    puts.clear();
  }
}

/**
 * Add coprocessor to values Map
 * @param specStr The Coprocessor specification all in in one String formatted so matches
 * {@link HConstants#CP_HTD_ATTR_VALUE_PATTERN}
 * @return Returns <code>this</code>
 */
private HTableDescriptor addCoprocessorToMap(final String specStr) {
  if (specStr == null) return this;
  // generate a coprocessor key
  int maxCoprocessorNumber = 0;
  Matcher keyMatcher;
  for (Map.Entry<ImmutableBytesWritable, ImmutableBytesWritable> e:
      this.values.entrySet()) {
    keyMatcher =
        HConstants.CP_HTD_ATTR_KEY_PATTERN.matcher(
            Bytes.toString(e.getKey().get()));
    if (!keyMatcher.matches()) {
      continue;
    }
    maxCoprocessorNumber = Math.max(Integer.parseInt(keyMatcher.group(1)), maxCoprocessorNumber);
  }
  maxCoprocessorNumber++;
  String key = "coprocessor$" + Integer.toString(maxCoprocessorNumber);
  this.values.put(new ImmutableBytesWritable(Bytes.toBytes(key)),
    new ImmutableBytesWritable(Bytes.toBytes(specStr)));
  return this;
}

static void verifyData(HRegion newReg, int startRow, int numRows, byte[] qf, byte[]... families)
    throws IOException {
  for (int i = startRow; i < startRow + numRows; i++) {
    byte[] row = Bytes.toBytes("" + i);
    Get get = new Get(row);
    for (byte[] family : families) {
      get.addColumn(family, qf);
    }
    Result result = newReg.get(get);
    Cell[] raw = result.rawCells();
    assertEquals(families.length, result.size());
    for (int j = 0; j < families.length; j++) {
      assertTrue(CellUtil.matchingRow(raw[j], row));
      assertTrue(CellUtil.matchingFamily(raw[j], families[j]));
      assertTrue(CellUtil.matchingQualifier(raw[j], qf));
    }
  }
}

@Override
public void preBulkLoadHFile(ObserverContext<RegionCoprocessorEnvironment> ctx,
                             List<Pair<byte[], String>> familyPaths) throws IOException {
  RegionCoprocessorEnvironment e = ctx.getEnvironment();
  assertNotNull(e);
  assertNotNull(e.getRegion());
  if (e.getRegion().getTableDesc().getTableName().equals(
      TestRegionObserverInterface.TEST_TABLE)) {
    assertNotNull(familyPaths);
    assertEquals(1,familyPaths.size());
    assertArrayEquals(familyPaths.get(0).getFirst(), TestRegionObserverInterface.A);
    String familyPath = familyPaths.get(0).getSecond();
    String familyName = Bytes.toString(TestRegionObserverInterface.A);
    assertEquals(familyPath.substring(familyPath.length()-familyName.length()-1),"/"+familyName);
  }
  ctPreBulkLoadHFile.incrementAndGet();
}

@Test
public void testFilterWithLessFailed() throws IOException {
    clean();
    {
        Put put = new Put(Bytes.toBytes(rowPrefix));
        put.addColumn(Bytes.toBytes(familyName), Bytes.toBytes("col_1"), Bytes.toBytes("col_1_var"));
        table.put(put);
    }

    {
        Get get = new Get(Bytes.toBytes(rowPrefix));
        Filter filter = new SingleColumnValueFilter(Bytes.toBytes(familyName), Bytes.toBytes("col_1"),
                CompareFilter.CompareOp.LESS, Bytes.toBytes("col_1_vaa"));
        get.setFilter(filter);
        Result result = table.get(get);
        assertTrue(result.getRow() == null);
    }
}

protected void checkModel(CellSetModel model) {
  Iterator<RowModel> rows = model.getRows().iterator();
  RowModel row = rows.next();
  assertTrue(Bytes.equals(ROW1, row.getKey()));
  Iterator<CellModel> cells = row.getCells().iterator();
  CellModel cell = cells.next();
  assertTrue(Bytes.equals(COLUMN1, cell.getColumn()));
  assertTrue(Bytes.equals(VALUE1, cell.getValue()));
  assertTrue(cell.hasUserTimestamp());
  assertEquals(cell.getTimestamp(), TIMESTAMP1);
  assertFalse(cells.hasNext());
  row = rows.next();
  assertTrue(Bytes.equals(ROW2, row.getKey()));
  cells = row.getCells().iterator();
  cell = cells.next();
  assertTrue(Bytes.equals(COLUMN2, cell.getColumn()));
  assertTrue(Bytes.equals(VALUE2, cell.getValue()));
  assertTrue(cell.hasUserTimestamp());
  assertEquals(cell.getTimestamp(), TIMESTAMP2);
  cell = cells.next();
  assertTrue(Bytes.equals(COLUMN3, cell.getColumn()));
  assertTrue(Bytes.equals(VALUE3, cell.getValue()));
  assertTrue(cell.hasUserTimestamp());
  assertEquals(cell.getTimestamp(), TIMESTAMP3);
  assertFalse(cells.hasNext());
}

@Test
public void testRowsAllExist() throws IOException {
    clean();
    String row = rowPrefix;
    putRow(row + 0, 1000);
    putRow(row + 1, 1000);

    Get get1 = new Get(Bytes.toBytes(row + 0));
    Get get2 = new Get(Bytes.toBytes(row + 1));
    List<Get> gets = new ArrayList<Get>();
    gets.add(get1);
    gets.add(get2);

    boolean[] exists = table.existsAll(gets);
    assertEquals(2, exists.length);
    assertTrue(exists[0]);
    assertTrue(exists[1]);
}

/**
 * 列限定符过滤器
 *
 * @param tableName  表名
 * @param columnName 列限定符
 * @param count      数量
 */
public void qualifierFilter(String tableName, String columnName, int count) {
    HBaseConfiguration hBaseConfiguration = new HBaseConfiguration();
    Table table = hBaseConfiguration.table(tableName);
    Scan scan = new Scan();
    //使用列族过滤器
    scan.setFilter(new QualifierFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(columnName))));//直接行健
    //scan.setFilter(new QualifierFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator("row.*")));//正则表达式
    //scan.setFilter(new QualifierFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator("row")));//字符串包含
    //scan.setFilter(new QualifierFilter(CompareFilter.CompareOp.EQUAL, new BinaryPrefixComparator("m".getBytes())));//字符串前缀
    scan.setCaching(10);
    scan.setBatch(10);
    try {
        ResultScanner scanner = table.getScanner(scan);
        Result[] results = scanner.next(count);
        HBaseResultUtil.print(results);
    } catch (IOException e) {
        e.printStackTrace();
    }
}

@Test
public void testFilterWithFilterIfMissingSucceeded() throws IOException {
    clean();
    {
        Put put = new Put(Bytes.toBytes(rowPrefix));
        put.addColumn(Bytes.toBytes(familyName), Bytes.toBytes("col_1"), Bytes.toBytes("col_1_var"));
        table.put(put);
    }

    {
        Get get = new Get(Bytes.toBytes(rowPrefix));
        SingleColumnValueFilter filter = new SingleColumnValueFilter(Bytes.toBytes(familyName), Bytes.toBytes("col_3"),
                CompareFilter.CompareOp.EQUAL, Bytes.toBytes("col_1_var"));
        filter.setFilterIfMissing(false);
        get.setFilter(filter);
        Result result = table.get(get);
        String value = Bytes.toString(result.getValue(Bytes.toBytes(familyName), Bytes.toBytes("col_1")));
        assertEquals("col_1_var", value);
    }
}

/**
 * 单列排除过滤器（返回的列 不包含参考列）
 *
 * @param tableName    表名
 * @param columnFamily 列族
 * @param qualifier    列限定符
 * @param columnValue  列值
 * @param count        数量
 */
public void SingleColumnValueExcludeFilter(String tableName, String columnFamily, String qualifier, String columnValue, int count) {
    HBaseConfiguration hBaseConfiguration = new HBaseConfiguration();
    Table table = hBaseConfiguration.table(tableName);
    Scan scan = new Scan();
    SingleColumnValueExcludeFilter singleColumnValueFilter = new SingleColumnValueExcludeFilter(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier), CompareFilter.CompareOp.EQUAL, Bytes.toBytes(columnValue));
    //singleColumnValueFilter.setFilterIfMissing(true);//当不存在这列的行 默认不过滤
    singleColumnValueFilter.setLatestVersionOnly(true);//获取最新版本
    scan.setFilter(singleColumnValueFilter);
    scan.setCaching(10);
    //scan.setBatch(10);
    try {
        ResultScanner scanner = table.getScanner(scan);
        Result[] results = scanner.next(count);
        HBaseResultUtil.print(results);
    } catch (IOException e) {
        e.printStackTrace();
    }
}

/** constructor */
public TestMinorCompaction() {
  super();

  // Set cache flush size to 1MB
  conf.setInt(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, 1024*1024);
  conf.setInt(HConstants.HREGION_MEMSTORE_BLOCK_MULTIPLIER, 100);
  compactionThreshold = conf.getInt("hbase.hstore.compactionThreshold", 3);

  firstRowBytes = START_KEY_BYTES;
  secondRowBytes = START_KEY_BYTES.clone();
  // Increment the least significant character so we get to next row.
  secondRowBytes[START_KEY_BYTES.length - 1]++;
  thirdRowBytes = START_KEY_BYTES.clone();
  thirdRowBytes[START_KEY_BYTES.length - 1] += 2;
  col1 = Bytes.toBytes("column1");
  col2 = Bytes.toBytes("column2");
}

protected static void parseUserResult(final String userName, final Result result,
    final UserQuotasVisitor visitor) throws IOException {
  Map<byte[], byte[]> familyMap = result.getFamilyMap(QUOTA_FAMILY_INFO);
  if (familyMap == null || familyMap.isEmpty()) return;

  for (Map.Entry<byte[], byte[]> entry : familyMap.entrySet()) {
    Quotas quotas = quotasFromData(entry.getValue());
    if (Bytes.startsWith(entry.getKey(), QUOTA_QUALIFIER_SETTINGS_PREFIX)) {
      String name = Bytes.toString(entry.getKey(), QUOTA_QUALIFIER_SETTINGS_PREFIX.length);
      if (name.charAt(name.length() - 1) == TableName.NAMESPACE_DELIM) {
        String namespace = name.substring(0, name.length() - 1);
        visitor.visitUserQuotas(userName, namespace, quotas);
      } else {
        TableName table = TableName.valueOf(name);
        visitor.visitUserQuotas(userName, table, quotas);
      }
    } else if (Bytes.equals(entry.getKey(), QUOTA_QUALIFIER_SETTINGS)) {
      visitor.visitUserQuotas(userName, quotas);
    }
  }
}

@Test (expected = EventDeliveryException.class)
public void testTimeOut() throws Exception {
  testUtility.createTable(tableName.getBytes(), columnFamily.getBytes());
  deleteTable = true;
  AsyncHBaseSink sink = new AsyncHBaseSink(testUtility.getConfiguration(), true, false);
  Configurables.configure(sink, ctx);
  Channel channel = new MemoryChannel();
  Configurables.configure(channel, ctx);
  sink.setChannel(channel);
  channel.start();
  sink.start();
  Transaction tx = channel.getTransaction();
  tx.begin();
  for (int i = 0; i < 3; i++) {
    Event e = EventBuilder.withBody(Bytes.toBytes(valBase + "-" + i));
    channel.put(e);
  }
  tx.commit();
  tx.close();
  Assert.assertFalse(sink.isConfNull());
  sink.process();
  Assert.fail();
}

public long getTimestamp(long ts) throws BadTsvLineException {
  // Return ts if HBASE_TS_KEY is not configured in column spec
  if (!hasTimestamp()) {
    return ts;
  }

  String timeStampStr = Bytes.toString(lineBytes,
      getColumnOffset(timestampKeyColumnIndex),
      getColumnLength(timestampKeyColumnIndex));
  try {
    return Long.parseLong(timeStampStr);
  } catch (NumberFormatException nfe) {
    // treat this record as bad record
    throw new BadTsvLineException("Invalid timestamp " + timeStampStr);
  }
}

/**
 * 往表中删除列族
 *
 * @param tableName  表名
 * @param familyName 列族名
 */
public void deleteColumn(String tableName, String familyName) {
    HBaseConfiguration hBaseConfiguration = new HBaseConfiguration();
    Admin admin = hBaseConfiguration.admin();
    TableName tb = TableName.valueOf(tableName);
    try {
        if (admin.tableExists(tb)) {
            admin.deleteColumn(tb, Bytes.toBytes(familyName));
        } else {
            log.info("表名【" + tableName + "】不存在");
        }

    } catch (IOException e) {
        log.error(e);
    } finally {
        hBaseConfiguration.close();
    }
}

@Test
public void testMultiRowMutation() throws Exception {
  LOG.info("Starting testMultiRowMutation");
  final TableName TABLENAME = TableName.valueOf("testMultiRowMutation");
  final byte [] ROW1 = Bytes.toBytes("testRow1");

  Table t = TEST_UTIL.createTable(TABLENAME, FAMILY);
  Put p = new Put(ROW);
  p.add(FAMILY, QUALIFIER, VALUE);
  MutationProto m1 = ProtobufUtil.toMutation(MutationType.PUT, p);

  p = new Put(ROW1);
  p.add(FAMILY, QUALIFIER, VALUE);
  MutationProto m2 = ProtobufUtil.toMutation(MutationType.PUT, p);

  MutateRowsRequest.Builder mrmBuilder = MutateRowsRequest.newBuilder();
  mrmBuilder.addMutationRequest(m1);
  mrmBuilder.addMutationRequest(m2);
  MutateRowsRequest mrm = mrmBuilder.build();
  CoprocessorRpcChannel channel = t.coprocessorService(ROW);
  MultiRowMutationService.BlockingInterface service =
     MultiRowMutationService.newBlockingStub(channel);
  service.mutateRows(null, mrm);
  Get g = new Get(ROW);
  Result r = t.get(g);
  assertEquals(0, Bytes.compareTo(VALUE, r.getValue(FAMILY, QUALIFIER)));
  g = new Get(ROW1);
  r = t.get(g);
  assertEquals(0, Bytes.compareTo(VALUE, r.getValue(FAMILY, QUALIFIER)));
}

private Result nextInternal() throws IOException {
    Result result = null;
    int indexOfResultToUse = -1;
    for (int i = 0; i < this.scanTasks.size(); ++i) {
        ScanTask scanTask = this.scanTasks.get(i);
        // fail fast in case of errors
        checkTask(scanTask);
        if (nextResults[i] == null) {
            try {
                nextResults[i] = scanTask.getResult();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return null;
            }
            if (nextResults[i] == null) {
                continue;
            }
        }
        if (result == null || Bytes.compareTo(keyDistributor.getOriginalKey(nextResults[i]
                        .getRow()),
                keyDistributor.getOriginalKey(result.getRow())) < 0) {
            result = nextResults[i];
            indexOfResultToUse = i;
        }
    }
    if (indexOfResultToUse >= 0) {
        nextResults[indexOfResultToUse] = null;
    }
    return result;
}

public List<Path> compact(CompactionRequest request, int targetCount, long targetSize,
    byte[] left, byte[] right, byte[] majorRangeFromRow, byte[] majorRangeToRow,
    CompactionThroughputController throughputController, User user) throws IOException {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Executing compaction with " + targetSize
        + " target file size, no more than " + targetCount + " files, in ["
        + Bytes.toString(left) + "] [" + Bytes.toString(right) + "] range");
  }
  StripeMultiFileWriter writer = new StripeMultiFileWriter.SizeMultiWriter(
      targetCount, targetSize, left, right);
  return compactInternal(writer, request, majorRangeFromRow, majorRangeToRow,
    throughputController, user);
}

@Test
public void testKeyOnlyFilter() throws IOException {
  String filterString = "KeyOnlyFilter()";
  doTestFilter(filterString, KeyOnlyFilter.class);

  String filterString2 = "KeyOnlyFilter ('') ";
  byte [] filterStringAsByteArray2 = Bytes.toBytes(filterString2);
  try {
    filter = f.parseFilterString(filterStringAsByteArray2);
    assertTrue(false);
  } catch (IllegalArgumentException e) {
    System.out.println(e.getMessage());
  }
}

@Test
public void testRepeatingKVs() throws Exception {
  List<KeyValue> kvs = Lists.newArrayList();
  for (int i = 0; i < 400; i++) {
    byte[] row = Bytes.toBytes("row" + (i % 10));
    byte[] fam = Bytes.toBytes("fam" + (i % 127));
    byte[] qual = Bytes.toBytes("qual" + (i % 128));
    kvs.add(new KeyValue(row, fam, qual, 12345L, VALUE));
  }

  runTestCycle(kvs);
}

@Test(expected = FlumeException.class)
public void testMissingTable() throws Exception {
  deleteTable = false;
  ctx.put("batchSize", "2");
  AsyncHBaseSink sink = new AsyncHBaseSink(testUtility.getConfiguration());
  Configurables.configure(sink, ctx);
  //Reset the context to a higher batchSize
  ctx.put("batchSize", "100");
  Channel channel = new MemoryChannel();
  Configurables.configure(channel, ctx);
  sink.setChannel(channel);
  sink.start();
  Transaction tx = channel.getTransaction();
  tx.begin();
  for (int i = 0; i < 3; i++) {
    Event e = EventBuilder.withBody(Bytes.toBytes(valBase + "-" + i));
    channel.put(e);
  }
  tx.commit();
  tx.close();
  sink.process();
  Assert.assertFalse(sink.isConfNull());
  HTable table = new HTable(testUtility.getConfiguration(), tableName);
  byte[][] results = getResults(table, 2);
  byte[] out;
  int found = 0;
  for (int i = 0; i < 2; i++) {
    for (int j = 0; j < 2; j++) {
      if (Arrays.equals(results[j], Bytes.toBytes(valBase + "-" + i))) {
        found++;
        break;
      }
    }
  }
  Assert.assertEquals(2, found);
  out = results[2];
  Assert.assertArrayEquals(Longs.toByteArray(2), out);
  sink.process();
  sink.stop();
}