使用MapReduce并行查询HBase中的行键列表

Question

我想在HBase中执行查询操作，以使用提供的行键列表提取记录。由于MapReduce中的Mappers是并行工作的，所以我想使用它。

行键的输入列表将在~100000的范围内，我已经为映射器创建了一个自定义的InputFormat，它将为每个映射器提供1000行键的列表以查询HBase表。这些查询的记录可能存在于 HBase 表中，也可能不存在，我只想返回那些存在的记录。

我看到了各种各样的例子，我发现hbase tablecan操作被执行以获取行键的范围，并且范围由startingRowKey和endingRowKey指定，但我只想查询提供的行键列表。

我如何用MapReduce实现这一点？欢迎任何帮助！

Answer 1

当您将行键列表传递给映射器时，您应该向HBase发出get请求。每个get返回所请求密钥的数据，如果密钥不存在，则不返回任何数据。

首先，您应该在映射器的 setup（） 方法中创建表实例：

private Table table;

@Override
protected void setup(Context context) throws IOException, InterruptedException {
    Configuration hbaseConfig = HBaseConfiguration.create();
    Connection conn = ConnectionFactory.createConnection(hbaseConfig);
    this.table = conn.getTable(TableName.valueOf("hbaseTable"));
}

然后，您可以借助get和Result实例，从每个键的< code>map()方法向HBase表发出< code>get请求:

String key = "keyString";
Get getValue = new Get(key.getBytes());

//add column family and column qualifier if you desire
getValue.addColumn("columnFamily".getBytes(), "columnQual".getBytes());

try {
    Result result = table.get(getValue);
    if (!table.exists(getValue)) {

        //requested key doesn't exist
        return;
    }

    // do what you want with result instance 
}

完成映射器的工作后，您需要在 cleanup（） 方法中关闭与表的连接;

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

此外，您可以自由地将get请求的结果传递给还原器，或者使用cleanup（）method组合它们。这只取决于你的目的。

Answer 2

您可以在映射器中使用这种方法，这对我来说很好，它将返回结果数组。

/**
     * Method getDetailRecords.
     * 
     * @param listOfRowKeys List<String>
     * @return Result[]
     * @throws IOException
     */
    private Result[] getDetailRecords(final List<String> listOfRowKeys) throws IOException {
        final HTableInterface table = HBaseConnection.getHTable(TBL_DETAIL);
        final List<Get> listOFGets = new ArrayList<Get>();
        Result[] results = null;
        try {
            for (final String rowkey : listOfRowKeys) {// prepare batch of get with row keys
   // System.err.println("get 'yourtablename', '" + saltIndexPrefix + rowkey + "'");
                final Get get = new Get(Bytes.toBytes(saltedRowKey(rowkey)));
                get.addColumn(COLUMN_FAMILY, Bytes.toBytes(yourcolumnname));
                listOFGets.add(get);
            }
            results = table.get(listOFGets);

        } finally {
            table.close();
        }
        return results;
    }