0

我想在不使用任何减速器的情况下读写 hbase。

我按照“The Apache HBase™ 参考指南”中的示例进行操作，但也有例外。

这是我的代码：

public class CreateHbaseIndex { 
static final String SRCTABLENAME="sourceTable";
static final String SRCCOLFAMILY="info";
static final String SRCCOL1="name";
static final String SRCCOL2="email";
static final String SRCCOL3="power";

static final String DSTTABLENAME="dstTable";
static final String DSTCOLNAME="index";
static final String DSTCOL1="key";
public static void main(String[] args) {
    System.out.println("CreateHbaseIndex Program starts!...");
    try {
        Configuration config = HBaseConfiguration.create();
        Scan scan = new Scan();
        scan.setCaching(500);
        scan.setCacheBlocks(false);
        scan.addColumn(Bytes.toBytes(SRCCOLFAMILY), Bytes.toBytes(SRCCOL1));//info:name
        HBaseAdmin admin = new HBaseAdmin(config);
        if (admin.tableExists(DSTTABLENAME)) {
            System.out.println("table Exists.");
        }
        else{
            HTableDescriptor tableDesc = new HTableDescriptor(DSTTABLENAME);
            tableDesc.addFamily(new HColumnDescriptor(DSTCOLNAME));
            admin.createTable(tableDesc);
            System.out.println("create table ok.");
        }
        Job job = new Job(config, "CreateHbaseIndex");
        job.setJarByClass(CreateHbaseIndex.class);
        TableMapReduceUtil.initTableMapperJob(
                SRCTABLENAME, // input HBase table name
                scan, // Scan instance to control CF and attribute selection
                HbaseMapper.class, // mapper
                ImmutableBytesWritable.class, // mapper output key
                Put.class, // mapper output value
                job);
        job.waitForCompletion(true);
    } catch (IOException e) {
        e.printStackTrace();
    } catch (InterruptedException e) {
        e.printStackTrace();
    } catch (ClassNotFoundException e) {
        e.printStackTrace();
    }
    System.out.println("Program ends!...");
}

public static class HbaseMapper extends TableMapper<ImmutableBytesWritable, Put> {
    private HTable dstHt;
    private Configuration dstConfig;
    @Override
    public void setup(Context context) throws IOException{
        dstConfig=HBaseConfiguration.create();
        dstHt = new HTable(dstConfig,SRCTABLENAME);
    }

    @Override
    public void map(ImmutableBytesWritable row, Result value, Context context) throws IOException, InterruptedException {
        // this is just copying the data from the source table...
        context.write(row, resultToPut(row,value));
    }

    private static Put resultToPut(ImmutableBytesWritable key, Result result) throws IOException {
        Put put = new Put(key.get());
        for (KeyValue kv : result.raw()) {
            put.add(kv);
        }
        return put;
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        dstHt.close();
        super.cleanup(context);
    }
}
}

顺便说一下，“souceTable”是这样的：

key  name    email
1    peter   a@a.com
2    sam     b@b.com

“dstTable”将是这样的：

key    value
peter  1
sam    2

我是这个领域的新手，需要你的帮助。谢谢~

4

2 回答 2

0

您是正确的，您不需要减速器来写入 HBase，但在某些情况下减速器可能会有所帮助。如果您正在创建索引，您可能会遇到两个映射器尝试写入同一行的情况。除非您小心确保它们写入不同的列限定符，否则由于竞争条件，您可能会用另一个更新覆盖一个更新。虽然 HBase 确实进行了行级锁定，但如果您的应用程序逻辑有问题，它也无济于事。

在没有看到您的异常的情况下，我猜您失败了，因为您试图将源表中的键值对写入不存在列族的索引表中。

于 2012-11-19T16:37:31.867 回答
0

在此代码中，您没有指定输出格式。您需要添加以下代码

    job.setOutputFormatClass(TableOutputFormat.class);

    job.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE,
            DSTTABLENAME);

此外，我们不应该在设置中创建新配置，我们需要使用上下文中的相同配置。

于 2013-11-18T13:24:37.153 回答