【问题标题】:Read/Write Hbase without reducer, exceptions不使用 reducer 读/写 Hbase,异常
【发布时间】:2012-11-16 16:31:30
【问题描述】:

我想在不使用任何reducer的情况下读写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

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

【问题讨论】:

  • reducer 是你写东西的地方。为什么不想在 reducer 中写入 hbase?
  • @ChrisGerken 我认为 mapper 可以完成我需要的所有事情。

标签: hadoop mapreduce hbase


【解决方案1】:

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

如果没有看到您的异常,我猜您失败了,因为您正试图将源表中的键值对写入索引表,其中列族不存在。

【讨论】:

    【解决方案2】:

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

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

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

    【讨论】:

      猜你喜欢
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 2017-09-02
      • 1970-01-01
      • 2014-06-18
      • 2016-01-12
      • 2019-02-21
      • 1970-01-01
      相关资源
      最近更新 更多