0

我正在使用 SingleColumnValueFilter 来返回我想要删除的行列表： 

SingleColumnValueFilter fileTimestampFilter = new SingleColumnValueFilter(
         Bytes.toBytes('a'),
         Bytes.toBytes('date'),
         CompareFilter.CompareOp.GREATER,
         Bytes.toBytes("20140101000000")
         );

然后我创建一个 Delete 对象并删除每一列。 

Delete delete = new Delete(Bytes.toBytes(rowKey));
delete.deleteColumn(Bytes.toBytes('a'), Bytes.toBytes('date'));
htable.delete(delete);

检索代码是

private List<String> getRecordsToDelete(long maxResultSize)
{
  ResultScanner rs = null;
  HTableInterface table = null;
  List<String> keyList = new ArrayList<String>();
  try
  {
    log.debug("Retrieving records");      
    HbaseConnection hbaseConnectionConfig = myConfig.getHbaseConnection();
    Configuration configuration = getHbaseConfiguration(hbaseConnectionConfig);
    table = new HTable(configuration, 'mytable');
    FilterList list = new FilterList(FilterList.Operator.MUST_PASS_ALL);
    Filter filter = HbaseDao.getFilter();
    list.addFilter(filter);
    list.addFilter(new PageFilter(maxResultSize));
    Scan scan = new Scan();
    scan.setFilter(list);
    //scan.setMaxResultSize(maxResultSize);
    //scan.setCaching(1);
    //scan.setCacheBlocks(false);
    //log.debug("Scan raw? = " + scan.isRaw());
    //scan.setRaw(false);
    rs = table.getScanner(scan);      
    Iterator<Result> iterator = rs.iterator();      
    while (iterator.hasNext())
    {        
      Result result = iterator.next();        
      String key = Bytes.toString(result.getRow());
      log.debug("**************** f key = " + key); //the same keys are always added here
      keyList.add(key);        
    }
    log.debug("Done processing retrieval of records to delete Size = " + keyList.size());
  }
  catch (Exception ex)
  {
    log.error("Unable to process retrieval of records.", ex);
  }
  finally
  {
    try
    {
      if (table !=  null)
      {
        table.close();
      }
      if (rs != null)
      {
        rs.close();
      }
    }
    catch (IOException ioEx)
    {
      //do nothing
      log.error(ioEx);
    }
  }
  return keyList;
}

此任务已安排，当它再次运行时，它正在检索相同的行。我知道 hbase 将行标记为删除，然后它们仅在主要压缩后才被物理删除。如果我在任务运行之间通过 hbase shell 查询该行，则该列肯定已被删除。为什么我的 Scan 在此任务的后续运行中返回相同的行？

提前致谢！

4

1 回答 1

1

它与主要压缩无关（默认情况下它们每 24 小时运行一次）。当您删除一行时，HBase 将忽略已删除的数据，直到最终删除（在major_compactions 上）。请注意，如果您没有启用自动刷新，则必须首先通过调用手动刷新客户端缓冲区htable.flushCommits()（默认情况下为 autoflush=on）。

您的问题可能是因为您只是删除a:date并且您的行有更多正在读取的列并且它们正在通过过滤器，因为如果没有值存在，这是默认行为。

如果要删除整行，只需删除delete.deleteColumn(Bytes.toBytes('a'), Bytes.toBytes('date'));即可删除行，而不仅仅是列。

如果您只想删除该a:date列，同时保持该行的其余部分不受影响，请设置 filterIfMissing 标志以避免行a:date == null通过（因为它已被删除）：filter.setFilterIfMissing(true);

或者为了获得最佳性能，仅将该列添加到扫描中，这将阻止读取其他列：scan.addColumn(Bytes.toBytes('a'), Bytes.toBytes('date'));

在旁注中，请注意list.addFilter(new PageFilter(maxResultSize));将从表的每个区域检索 maxResultSize 结果，您必须在 keyList 达到 maxResultSize 时通过打破它来手动实现迭代器内的限制。

还有一个提示，当出于调试目的而记录时，请始终记录完整的结果，以便准确查看其中的内容。

于 2014-12-31T06:49:15.760 回答