HBase 中加盐（Salting）之后的表如何读取：协处理器篇

Size: px

Start display at page:

Download "HBase 中加盐（Salting）之后的表如何读取：协处理器篇"

匾赖
4 years ago
Views:

1 在 HBase Rowkey 设计指南文章中, 我们介绍了避免数据热点的三种比较常见方法 : 加盐 - Salting 哈希 - Hashing 反转 - Reversing 其中在加盐 (Salting) 的方法里面是这么描述的 : 给 Rowkey 分配一个随机前缀以使得它和之前排序不同但是在 Rowkey 前面加了随机前缀, 那么我们怎么将这些数据读出来呢? 我将分三篇文章来介绍如何读取加盐之后的表, 其中每篇文章提供一种方法, 主要包括 : 使用协处理器读取加盐的表使用 Spark 读取加盐的表使用 MapReduce 读取加盐的表关于协处理器的入门及实战, 请参见这里本文使用的各组件版本 :hadoop-2.7.7,hbase-2.0.4,jdk1.8.0_201 测试数据生成在介绍如何查询数据之前, 我们先创建一张名为 iteblog 的 HBase 表, 用于测试为了数据均匀和介绍的方便, 这里使用了预分区, 并设置了 27 个分区, 如下 : hbase(main):002:0> create 'iteblog', 'f', SPLITS => ['A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L', 'M', 'N', 'O', 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X', 'Y', 'Z'] 0 row(s) in seconds 然后我们使用下面方法生成了条测试数据 RowKey 的形式为 UID + 当前数据生成时间戳 ; 由于 UID 的长度为 4, 所以条数据会存在大量的 UID 一样的数据, 所以我们使用加盐方法将这些数据均匀分散到上述 27 个 Region 里面 ( 注意, 其实第一个 Region 其实没数据 ) 具体代码如下 : package com.iteblog.data; import org.apache.hadoop.conf.configuration; import org.apache.hadoop.hbase.hbaseconfiguration; import org.apache.hadoop.hbase.hconstants; import org.apache.hadoop.hbase.tablename; 1 / 12

2 import org.apache.hadoop.hbase.client.*; import org.apache.hadoop.hbase.util.bytes; import java.io.ioexception; import java.util.arraylist; import java.util.list; import java.util.random; import java.util.uuid; public class HBaseDataGenerator { private static byte[] FAMILY = "f".getbytes(); private static byte[] QUALIFIER_UUID = "uuid".getbytes(); private static byte[] QUALIFIER_AGE = "age".getbytes(); private static char generateletter() { return (char) (Math.random() * 26 + 'A'); private static long generateuid(int n) { return (long) (Math.random() * 9 * Math.pow(10, n - 1)) + (long) Math.pow(10, n - 1); public static void main(string[] args) throws IOException { BufferedMutatorParams bmp = new BufferedMutatorParams(TableName.valueOf("iteblog ")); bmp.writebuffersize(1024 * 1024 * 24); Configuration conf = HBaseConfiguration.create(); conf.set(hconstants.zookeeper_quorum, " Connection connection = ConnectionFactory.createConnection(conf); BufferedMutator bufferedmutator = connection.getbufferedmutator(bmp); int BATCH_SIZE = 1000; int COUNTS = ; int count = 0; List<Put> putlist = new ArrayList<>(); for (int i = 0; i < COUNTS; i++) { String rowkey = generateletter() + "-" + generateuid(4) + "-" + System.currentTimeMillis(); Put put = new Put(Bytes.toBytes(rowKey)); byte[] uuidbytes = UUID.randomUUID().toString().substring(0, 23).getBytes(); put.addcolumn(family, QUALIFIER_UUID, uuidbytes); 2 / 12

3 )); put.addcolumn(family, QUALIFIER_AGE, Bytes.toBytes("" + new Random().nextInt(100) putlist.add(put); count++; if (count % BATCH_SIZE == 0) { bufferedmutator.mutate(putlist); bufferedmutator.flush(); putlist.clear(); System.out.println(count); if (putlist.size() > 0) { bufferedmutator.mutate(putlist); bufferedmutator.flush(); putlist.clear(); 运行完上面代码之后, 会生成条数据 ( 注意, 这里其实不严谨, 因为 Rowkey 设计问题, 可能会导致重复的 Rowkey 生成, 所以实际情况下可能没有条数据 ) 我们 limit 10 条数据看下长成什么样 : hbase(main):001:0> scan 'iteblog', {'LIMIT'=>10 ROW COLUMN+CELL A column=f:age, timestamp= , value=54 A column=f:uuid, timestamp= , value=e9b10a9f fd-bd01 A column=f:age, timestamp= , value=4 A column=f:uuid, timestamp= , value=181aa91e-5f1d -454c-959c A column=f:age, timestamp= , value=33 A column=f:uuid, timestamp= , value=19aad8d3-621a -473c-8f9f A column=f:age, timestamp= , value=64 A column=f:uuid, timestamp= , value=c6712a0d d5-865b A column=f:age, timestamp= , value=96 A column=f:uuid, timestamp= , value=4bf05d10-bb4d -43e / 12

4 A A a3-8f96 A A d95 A A b68 A A c f-89c0 A A e-bd0c 10 row(s) Took seconds column=f:age, timestamp= , value=8 column=f:uuid, timestamp= , value=f04ba835-d8accolumn=f:age, timestamp= , value=84 column=f:uuid, timestamp= , value=99d6c989-afb5- column=f:age, timestamp= , value=21 column=f:uuid, timestamp= , value=8c1ff1b6-b97ccolumn=f:age, timestamp= , value=2 column=f:uuid, timestamp= , value=e240aa0f- column=f:age, timestamp= , value=6 column=f:uuid, timestamp= , value=e8df15c9-02fa- 使用协处理器查询加盐之后的表现在有数据了, 我们需要查询所有 UID = 1000 的用户所有历史数据, 那么如何查呢? 我们知道 UID = 1000 的用户数据是均匀放到上述的 27 个 Region 里面的, 因为经过加盐了, 所以这些数据前缀都是类似于 A-,B-,C- 等开头的其次我们需要知道, 每个 Region 其实是有 Start Key 和 End Key 的, 这些 Start Key 和 End Key 其实就是我们创建 iteblog 表指定的如果你看了 HBase 协处理器入门及实战这篇文章, 你就知道协处理器的代码其实是在每个 Region 里面执行的 ; 而这些代码在 Region 里面执行的时候是可以拿到当前 Region 的信息, 包括了 Start Key 和 End Key, 所以其实我们可以将拿到的 Start Key 信息和查询的 UID 进行拼接, 这样就可以查询我们要的数据协处理器处理篇就是基于这样的思想来查询加盐之后的数据的定义 proto 文件为什么需要定义这个请参见 HBase 协处理器入门及实战这篇文章因为我们查询的时候需要传入查询的参数, 比如 tablename StartKey EndKey 以及是否加盐等标记 ; 同时当查询到结果的时候, 我们还需要将数据返回, 所以我们定义的 proto 文件如下 : option java_package = "com.iteblog.data.coprocessor.generated"; option java_outer_classname = "DataQueryProtos"; option java_generic_services = true; option java_generate_equals_and_hash = true; option optimize_for = SPEED; 4 / 12

5 message DataQueryRequest { optional string tablename = 1; optional string startrow = 2; optional string endrow = 3; optional bool incluedend = 4; optional bool issalting = 5; message DataQueryResponse { message Cell{ required bytes value = 1; required bytes family = 2; required bytes qualifier = 3; required bytes row = 4; required int64 timestamp = 5; message Row{ optional bytes rowkey = 1; repeated Cell celllist = 2; repeated Row rowlist = 1; service QueryDataService{ rpc querybystartrowandendrow(dataqueryrequest) returns (DataQueryResponse); 然后我们使用 protobuf-maven-plugin 插件将上面的 proto 生成 java 类, 具体如何操作参见在 IDEA 中使用 Maven 编译 proto 文件我们将生成的 DataQueryProtos.java 类拷贝到 com.iteblog.data.coprocessor.generated 包里面编写协处理器代码有了请求和返回的类, 现在我们需要编写协处理器的处理代码了, 结合上面的分析, 协处理器的代码实现如下 : package com.iteblog.data.coprocessor; import com.google.protobuf.bytestring; 5 / 12

6 import com.google.protobuf.rpccallback; import com.google.protobuf.rpccontroller; import com.google.protobuf.service; import com.iteblog.data.coprocessor.generated.dataqueryprotos.querydataservice; import com.iteblog.data.coprocessor.generated.dataqueryprotos.dataqueryrequest; import com.iteblog.data.coprocessor.generated.dataqueryprotos.dataqueryresponse; import org.apache.hadoop.hbase.cell; import org.apache.hadoop.hbase.coprocessorenvironment; import org.apache.hadoop.hbase.client.get; import org.apache.hadoop.hbase.client.result; import org.apache.hadoop.hbase.client.scan; import org.apache.hadoop.hbase.coprocessor.coprocessorexception; import org.apache.hadoop.hbase.coprocessor.regioncoprocessor; import org.apache.hadoop.hbase.coprocessor.regioncoprocessorenvironment; import org.apache.hadoop.hbase.regionserver.internalscanner; import org.apache.hadoop.hbase.shaded.protobuf.responseconverter; import org.apache.hadoop.hbase.util.bytes; import java.io.ioexception; import java.util.arraylist; import java.util.collections; import java.util.list; public class SlatTableDataSearch extends QueryDataService implements RegionCoprocessor { private RegionCoprocessorEnvironment env; public Iterable<Service> getservices() { return public void querybystartrowandendrow(rpccontroller controller, DataQueryRequest request, RpcCallback<DataQueryResponse> done) { DataQueryResponse response = null; String startrow = request.getstartrow(); String endrow = request.getendrow(); String regionstartkey = Bytes.toString(this.env.getRegion().getRegionInfo().getStartKey()); if (request.getissalting()) { String startsalt = null; if (null!= regionstartkey && regionstartkey.length()!= 0) { startsalt = regionstartkey; if (null!= startsalt && null!= startrow) { 6 / 12

7 startrow = startsalt + "-" + startrow; endrow = startsalt + "-" + endrow; Scan scan = new Scan(); if (null!= startrow) { scan.withstartrow(bytes.tobytes(startrow)); if (null!= endrow) { scan.withstoprow(bytes.tobytes(endrow), request.getincluedend()); try (InternalScanner scanner = this.env.getregion().getscanner(scan)) { List<Cell> results = new ArrayList<>(); boolean hasmore; DataQueryResponse.Builder responsebuilder = DataQueryResponse.newBuilder(); do { hasmore = scanner.next(results); DataQueryResponse.Row.Builder rowbuilder = DataQueryResponse.Row.newBuilder( ); if (results.size() > 0) { Cell cell = results.get(0); rowbuilder.setrowkey(bytestring.copyfrom(cell.getrowarray(), cell.getrowoffset(), cell.getrowlength())); for (Cell kv : results) { buildcell(rowbuilder, kv); responsebuilder.addrowlist(rowbuilder); results.clear(); while (hasmore); response = responsebuilder.build(); catch (IOException e) { ResponseConverter.setControllerException(controller, e); done.run(response); private void buildcell(dataqueryresponse.row.builder rowbuilder, Cell kv) { DataQueryResponse.Cell.Builder cellbuilder = DataQueryResponse.Cell.newBuilder(); 7 / 12

8 cellbuilder.setfamily(bytestring.copyfrom(kv.getfamilyarray(), kv.getfamilyoffset(), kv.get FamilyLength())); cellbuilder.setqualifier(bytestring.copyfrom(kv.getqualifierarray(), kv.getqualifieroffset(), kv.getqualifierlength())); cellbuilder.setrow(bytestring.copyfrom(kv.getrowarray(), kv.getrowoffset(), kv.getrowle ngth())); cellbuilder.setvalue(bytestring.copyfrom(kv.getvaluearray(), kv.getvalueoffset(), kv.getval uelength())); cellbuilder.settimestamp(kv.gettimestamp()); rowbuilder.addcelllist(cellbuilder); /** * Stores a reference to the coprocessor environment provided by the * {@link org.apache.hadoop.hbase.regionserver.regioncoprocessorhost from the region where this * coprocessor is loaded. Since this is a coprocessor endpoint, it always expects to be loade d * on a table region, so always expects this to be an instance of * {@link RegionCoprocessorEnvironment. * env the environment provided by the coprocessor host IOException if the provided environment is not an instance of * {@code RegionCoprocessorEnvironment public void start(coprocessorenvironment env) throws IOException { if (env instanceof RegionCoprocessorEnvironment) { this.env = (RegionCoprocessorEnvironment) env; else { throw new CoprocessorException("Must be loaded on a table public void stop(coprocessorenvironment env) { // nothing to do 大家可以看到, 这里面的代码框架和 HBase 协处理器入门及实战里面介绍的 HBase 提供的 RowCountEndpoint 示例代码很类似主要逻辑在 querybystartrowandendrow 函数实现里面我们通过 DataQueryRequest 拿到客户端查询的表,StartKey 和 EndKey 8 / 12

9 等数据通过 this.env.getregion().getregioninfo().getstartkey() 可以拿到当前 Region 的 StartKey, 然后再和客户端传进来的 StartKey 和 EndKey 进行拼接就可以拿到完整的 Rowkey 前缀剩下的查询就是正常的 HBase Scan 代码了现在我们将 SlatTableDataSearch 类进行编译打包, 并部署到 HBase 表里面去, 具体如何部署参见 HBase 协处理器入门及实战协处理器客户端代码编写到这里, 我们的协处理器服务器端的代码和部署已经完成了, 现在我们需要编写协处理器客户端代码其实也很简单, 如下 : package com.iteblog.data; import com.iteblog.data.coprocessor.generated.dataqueryprotos.querydataservice; import com.iteblog.data.coprocessor.generated.dataqueryprotos.dataqueryrequest; import com.iteblog.data.coprocessor.generated.dataqueryprotos.dataqueryresponse; import com.iteblog.data.coprocessor.generated.dataqueryprotos.dataqueryresponse.*; import org.apache.hadoop.conf.configuration; import org.apache.hadoop.hbase.hbaseconfiguration; import org.apache.hadoop.hbase.tablename; import org.apache.hadoop.hbase.client.connection; import org.apache.hadoop.hbase.client.connectionfactory; import org.apache.hadoop.hbase.client.htable; import org.apache.hadoop.hbase.ipc.coprocessorrpcutils.blockingrpccallback; import org.apache.hadoop.hbase.ipc.serverrpccontroller; import java.util.linkedlist; import java.util.list; import java.util.map; public class DataQuery { private static Configuration conf = null; static { conf = HBaseConfiguration.create(); conf.set("hbase.zookeeper.quorum", " static List<Row> querybystartrowandstoprow(string tablename, String startrow, String stoprow, boolean isincludeend, boolean issalting) { final DataQueryRequest.Builder requestbuilder = DataQueryRequest.newBuilder(); requestbuilder.settablename(tablename); requestbuilder.setstartrow(startrow); 9 / 12

10 requestbuilder.setendrow(stoprow); requestbuilder.setincluedend(isincludeend); requestbuilder.setissalting(issalting); try { Connection connection = ConnectionFactory.createConnection(conf); HTable table = (HTable) connection.gettable(tablename.valueof(tablename)); Map<byte[], List<Row>> result = table.coprocessorservice(querydataservice.class, null, null, counter -> { ServerRpcController controller = new ServerRpcController(); BlockingRpcCallback<DataQueryResponse> call = new BlockingRpcCallback<>(); counter.querybystartrowandendrow(controller, requestbuilder.build(), call); DataQueryResponse response = call.get(); if (controller.failedonexception()) { throw controller.getfailedon(); return response.getrowlistlist(); ); List<Row> list = new LinkedList<>(); for (Map.Entry<byte[], List<Row>> entry : result.entryset()) { if (null!= entry.getkey()) { list.addall(entry.getvalue()); return list; catch (Throwable e) { e.printstacktrace(); return null; public static void main(string[] args) { List<Row> rows = querybystartrowandstoprow("iteblog", "1000", "1001", false, true); if (null!= rows) { System.out.println(rows.size()); for (DataQueryResponse.Row row : rows) { List<DataQueryResponse.Cell> celllistlist = row.getcelllistlist(); for (DataQueryResponse.Cell cell : celllistlist) { System.out.println(row.getRowKey().toStringUtf8() + " \t " + "column=" + cell.getfamily().tostringutf8() + ":" + cell.getqualifier().tostringutf8() + ", " + "timestamp=" + cell.gettimestamp() + ", " + 10 / 12

11 "value=" + cell.getvalue().tostringutf8()); 我们运行上面的代码, 可以得到如下的输出 : A column=f:age, timestamp= , value=54 A column=f:uuid, timestamp= , value=e9b10a9f fd-bd01 A column=f:age, timestamp= , value=4 A column=f:uuid, timestamp= , value=181aa91e-5f1d-4 54c-959c A column=f:age, timestamp= , value=33 A column=f:uuid, timestamp= , value=19aad8d3-621a-4 73c-8f9f B column=f:age, timestamp= , value=1 B column=f:uuid, timestamp= , value=cf720efe-2ad2-48 d6-81b8 B column=f:age, timestamp= , value=7 B column=f:uuid, timestamp= , value=8a e cb-adfe hbase(main):020:0> scan 'iteblog', {STARTROW => 'A-1000', ENDROW => 'A-1001' ROW COLUMN+CELL A column=f:age, timestamp= , value=54 A column=f:uuid, timestamp= , value=e9b10a9f fd-bd01 A column=f:age, timestamp= , value=4 A column=f:uuid, timestamp= , value=181aa91e-5f1d -454c-959c A column=f:age, timestamp= , value=33 A column=f:uuid, timestamp= , value=19aad8d3-621 a-473c-8f9f 3 row(s) Took seconds 可以看到, 和我们使用 HBase Shell 输出的一致, 而且我们还把所有的 UID = / 12

12 Powered by TCPDF ( 的数据拿到了好了, 到这里, 使用协处理器查询 HBase 加盐之后的表已经算完成了, 明天我将介绍使用 Spark 如何查询加盐之后的表本博客文章除特别声明, 全部都是原创! 转载本文请加上 : 转载自过往记忆 ( 本文链接 : () 12 / 12

新・解きながら学ぶJava

新・解きながら学ぶJava 481! 41, 74!= 40, 270 " 4 % 23, 25 %% 121 %c 425 %d 121 %o 121 %x 121 & 199 && 48 ' 81, 425 ( ) 14, 17 ( ) 128 ( ) 183 * 23 */ 3, 390 ++ 79 ++ 80 += 93 + 22 + 23 + 279 + 14 + 124 + 7, 148, 16 -- 79 --