在 HBase 单元中保存多个版本

Question

我是 HBase 的新手。我试图在 HBase 的一个单元格中保存多个版本，但我只获得最后保存的值。我尝试了以下两个命令来检索多个保存的版本： get 'Dummy1','abc', {COLUMN=>'backward:first', VERSIONS=>12} 和 scan 'Dummy1', {VERSIONS=>12} 两者都返回如下输出：

ROW                   COLUMN+CELL                                               
 abc                  column=backward:first, timestamp=1422722312845, value=rrb 

0.0150 秒内 1 行 输入文件如下：

abc xyz kkk
abc qwe asd
abc anf rrb

HBase中建表的代码如下：

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;

public class HBaseTableCreator {

  public static void main(String[] args) throws Exception {

      HBaseConfiguration conf = new HBaseConfiguration();
      conf.set("hbase.master","localhost:60000");

      HBaseAdmin hbase = new HBaseAdmin(conf);
      HTableDescriptor desc = new HTableDescriptor("Dummy");
      HColumnDescriptor meta = new HColumnDescriptor("backward".getBytes());
      meta.setMaxVersions(Integer.MAX_VALUE);
      HColumnDescriptor prefix = new HColumnDescriptor("forward".getBytes());
      prefix.setMaxVersions(Integer.MAX_VALUE);
      desc.addFamily(meta);
      desc.addFamily(prefix);
      hbase.createTable(desc);

 }

}

转储HBase数据的代码如下： 主类： 导入 java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;


public class TestMain {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException 
    {
        // TODO Auto-generated method stub
        Configuration conf=new Configuration();
        //HTable hTable = new HTable(conf, args[3]);  
        String[] otherArgs=new GenericOptionsParser(conf,args).getRemainingArgs();
        if(otherArgs.length!=2)
        {
            System.err.println("Usage: wordcount <in> <out>");
            System.exit(2);
        }
        Job job=new Job(conf,"HBase dummy dump");
        job.setJarByClass(TestMain.class);
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class); 
        job.setMapperClass(TestMapper.class);
        TableMapReduceUtil.initTableReducerJob("Dummy", null, job);
        //job.setOutputKeyClass(NullWritable.class);
        //job.setOutputValueClass(Text.class);
        job.setNumReduceTasks(0);
        //job.setOutputKeyClass(Text.class);
        //job.setOutputValueClass(Text.class);
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
        //HFileOutputFormat.configureIncrementalLoad(job, hTable);
        System.exit(job.waitForCompletion(true)?0:1);
    }
}

映射器类：

import java.io.IOException;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Mapper;

public class TestMapper extends Mapper <LongWritable, Text, Text, Put>{
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 

        String line=value.toString();
        String[] l=line.split("\\s+");
        for(int i=1;i<l.length;i++)
        {
            Put HPut = new Put(l[0].getBytes());
            HPut.add("backward".getBytes(),"first".getBytes(),l[i].getBytes());
            context.write(new Text(l[0]),HPut);
        }
    }
}

请告诉我哪里出错了。

Answer 1

您的问题是您的写入会自动进行批处理，并且它们在作业结束时被刷新（当表关闭时），可能导致每个 put 操作具有完全相同的时间戳，并且它们基本上是覆盖自己（编写一个与另一个版本具有相同时间戳的版本会覆盖该版本而不是插入一个新版本）。

解决问题的第一种方法可能是自己提供时间戳Put HPut = new Put(l[0].getBytes(), System.currentTimeMillis());，但您可能会遇到同样的问题，因为操作非常快，以至于很多 put 将具有相同的时间戳。

这就是我要解决的问题：

1- 停止使用 TableMapReduceUtil.initTableReducerJob，转而使用自定义 reducer 来处理对 hbase 表的写入。

2- 修改映射器以将每行的所有值写入上下文，以便将它们分组为可迭代并传递给减速器（即：abc, xyz kkk qwe asd anf rrb）

3- 实现我自己的 reducer，有点像这样伪代码：

Define myHTable
setup() {
  Instantiate myHtable
  Disable myHtable autoflush to prevent puts from being automatically flushed
  Set myHtable write buffer to at least 2MB
}
reduce(rowkey, results) {
  baseTimestamp = current time in milliseconds
  Iterate results {
     Instantiate put with rowkey ++baseTimestamp
     Add result to put
     Send put to myHTable
  }
}
cleanup() {
  Flush commits for myHTable
  Close myHTable
}

那样的话，每个版本之间总会有 1ms 的间隔，你唯一需要注意的是，如果你有大量的版本并且多次运行同一个作业，新作业的时间戳可能与前一个的时间戳重叠，如果您预计版本少于 30k，则不必担心，因为每个作业距离下一个作业至少 30 秒...

无论如何，请注意，不建议拥有超过一百个版本 (http://hbase.apache.org/book.html#versions)，如果您需要更多版本，最好采用不带任何版本。

抱歉，奇怪的格式，这是让伪代码很好地显示的唯一方法。