在 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 

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

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 中转储数据的代码如下所示：主类： import 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 HPut = new Put(l[0].getBytes(), System.currentTimeMillis());但您可能会遇到同样的问题，因为操作速度如此之快，以至于很多放置将具有相同的时间戳。

这就是我会做的来克服这个问题：

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

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

）

3-实现我自己的化简器，使其工作得有点像这个伪代码：

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
}

这样，每个版本之间将始终有 1 毫秒，您唯一需要注意的是，如果您拥有大量版本并多次运行同一作业，则新作业的时间戳可能会与前一个作业的时间戳重叠，如果您预计少于 30k 版本，则不必担心，因为每个作业与下一个作业至少相距 30 秒一。。。

无论如何，请注意，不建议拥有超过一百个版本（http://hbase.apache.org/book.html#versions），如果您需要更多版本，则更明智的做法是采用一种没有版本的高方法（包含键+时间戳的复合行键）。

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