云数据库 Hbase 操作指南 产品文档
版权声明 2015-2016 腾讯云版权所有 本文档著作权归腾讯云单独所有, 未经腾讯云事先书面许可, 任何主体不得以任何形式复制 修改 抄袭 传 播全部或部分本文档内容 商标声明 及其它腾讯云服务相关的商标均为腾讯云计算 ( 北京 ) 有限责任公司及其关联公司所有 本文档涉及的第三方 主体的商标, 依法由权利人所有 服务声明 本文档意在向客户介绍腾讯云全部或部分产品 服务的当时的整体概况, 部分产品 服务的内容可能有所调整 您所购买的腾讯云产品 服务的种类 服务标准等应由您与腾讯云之间的商业合同约定, 除非双方另有约定, 否则, 腾讯云对本文档内容不做任何明示或模式的承诺或保证 第 2 页共 41 页
文档目录 文档声明... 2 Step1. 连接和访问... 4 Step2. 替换客户端 jar 包... 6 Step3. 连接腾讯云参数设置... 17 Step4. 操作示例代码... 18 MapReduce 示例代码... 25 endpoint 协处理器示例代码... 30 observer 协处理器示例... 40 第 3 页共 41 页
Step1. 连接和访问 获取连接 IP 1) 申请资格 点击进入产品介绍页申请测试资格 2) 获取实例 ID 和 IP 信息 进入腾讯云 Hbase 管理中心, 可以查看实例 ID, 并获取连接 Hbase 的一个或多个 IP 和端口 : 通过 SHELL 访问 首先下载 Hbase 客户端软件, 然后解压到云主机, 然后修改 conf 下的 hbase-site.xml 添加如下配置项目 : <configuration> <property> <name>hbase.zookeeper.quorum</name> <value>( 腾讯云提供的连接地址和端口, 管理控制台可查 )</value> </property> <property> <name> chbase.tencent.enable </name> <value> true</value> </property> </configuration> 然后执行 bin/hbase shell, 可以进入 Hbase 命令终端 : 第 4 页共 41 页
第 5 页共 41 页
Step2. 替换客户端 jar 包 替换客户端 jar 包 连接腾讯云 Hbase 服务, 需要使用我们提供的 jar 包来替换社区版本的部分 jar 包, 并设置相应的参数 1). 如您使用 maven 管理 jar 包, 可参考如下 pom 设置来拉取和替换 jar 包 : <properties> <project.build.sourceencoding>utf-8</project.build.sourceencoding> <hadoop-two.version>2.6.4</hadoop-two.version> </properties> <dependencies> <!-- 导入社区 jar 包 --> <groupid>org.apache.hbase</groupid> <artifactid>hbase-client</artifactid> <version>1.1.3</version> <type>pom</type> <scope>provided</scope> <exclusions> <artifactid>hadoop-annotations</artifactid> <artifactid>hadoop-auth</artifactid> <artifactid>hadoop-client</artifactid> 第 6 页共 41 页
<artifactid>hadoop-common</artifactid> <artifactid>hadoop-hdfs</artifactid> <artifactid>hadoop-mapreduce-client-app</artifactid> <artifactid>hadoop-mapreduce-client-common</artifactid> <artifactid>hadoop-mapreduce-client-core</artifactid> <artifactid>hadoop-mapreduce-client-jobclient</artifactid> <artifactid>hadoop-mapreduce-client-shuffle</artifactid> <artifactid>hadoop-yarn-api</artifactid> 第 7 页共 41 页
<artifactid>hadoop-yarn-client</artifactid> <artifactid>hadoop-yarn-common</artifactid> <artifactid>hadoop-yarn-server-common</artifactid> </exclusions> <groupid>org.apache.hbase</groupid> <artifactid>hbase-server</artifactid> <version>1.1.3</version> <type>pom</type> <scope>provided</scope> <exclusions> <artifactid>hadoop-annotations</artifactid> <artifactid>hadoop-auth</artifactid> <artifactid>hadoop-client</artifactid> <artifactid>hadoop-common</artifactid> 第 8 页共 41 页
<artifactid>hadoop-hdfs</artifactid> <artifactid>hadoop-mapreduce-client-app</artifactid> <artifactid>hadoop-mapreduce-client-common</artifactid> <artifactid>hadoop-mapreduce-client-core</artifactid> <artifactid>hadoop-mapreduce-client-jobclient</artifactid> <artifactid>hadoop-mapreduce-client-shuffle</artifactid> <artifactid>hadoop-yarn-api</artifactid> <artifactid>hadoop-yarn-client</artifactid> 第 9 页共 41 页
<artifactid>hadoop-yarn-common</artifactid> <artifactid>hadoop-yarn-server-common</artifactid> </exclusions> <groupid>com.google.code.gson</groupid> <artifactid>gson</artifactid> <version>2.5</version> <groupid>commons-configuration</groupid> <artifactid>commons-configuration</artifactid> <version>1.6</version> <groupid>org.apache.httpcomponents</groupid> <artifactid>httpcore</artifactid> <version>4.2.4</version> <groupid>org.apache.httpcomponents</groupid> <artifactid>httpclient</artifactid> <version>4.2.5</version> <groupid>org.htrace</groupid> <artifactid>htrace-core</artifactid> <version>3.0.4</version> 第 10 页共 41 页
<groupid>org.apache.avro</groupid> <artifactid>avro</artifactid> <version>1.7.4</version> <!-- 导入腾讯云提供的 jar 包, 注意修改 jar 包路径 --> <artifactid>hadoop-annotations</artifactid> <version>2.6.4</version> <scope>system</scope> <systempath>${basedir/hadoop/hadoop-annotations-2.6.4.jar</systempath> <artifactid>hadoop-auth</artifactid> <version>2.6.4</version> <scope>system</scope> <systempath>${basedir/hadoop/hadoop-auth-2.6.4.jar</systempath> <artifactid>hadoop-client</artifactid> <version>2.6.4</version> <scope>system</scope> <systempath>${basedir/hadoop/hadoop-client-2.6.4.jar</systempath> <artifactid>hadoop-common</artifactid> <version>2.6.4</version> <scope>system</scope> <systempath>${basedir/hadoop/hadoop-common-2.6.4.jar</systempath> 第 11 页共 41 页
<artifactid>hadoop-hdfs</artifactid> <version>2.6.4</version> <scope>system</scope> <systempath>${basedir/hadoop/hadoop-hdfs-2.6.4.jar</systempath> <artifactid>hadoop-mapreduce-client-app</artifactid> <version>2.6.4</version> <scope>system</scope> <systempath>${basedir/hadoop/hadoop-mapreduce-client-app-2.6.4.jar</systempath> <artifactid>hadoop-mapreduce-client-common</artifactid> <version>2.6.4</version> <scope>system</scope> <systempath> ${basedir/hadoop/hadoop-mapreduce-client-common-2.6.4.jar</systempath> <artifactid>hadoop-mapreduce-client-core</artifactid> <version>2.6.4</version> <scope>system</scope> 第 12 页共 41 页
<systempath> ${basedir/hadoop/hadoop-mapreduce-client-core-2.6.4.jar</systempath> <artifactid>hadoop-mapreduce-client-jobclient</artifactid> <version>2.6.4</version> <scope>system</scope> <systempath> ${basedir/hadoop/hadoop-mapreduce-client-jobclient-2.6.4.jar</systempath> <artifactid>hadoop-mapreduce-client-shuffle</artifactid> <version>2.6.4</version> <scope>system</scope> <systempath> ${basedir/hadoop/hadoop-mapreduce-client-shuffle-2.6.4.jar</systempath> <artifactid>hadoop-yarn-api</artifactid> <version>2.6.4</version> <scope>system</scope> <systempath>${basedir/hadoop/hadoop-yarn-api-2.6.4.jar</systempath> 第 13 页共 41 页
<artifactid>hadoop-yarn-client</artifactid> <version>2.6.4</version> <scope>system</scope> <systempath>${basedir/hadoop/hadoop-yarn-client-2.6.4.jar</systempath> <artifactid>hadoop-yarn-common</artifactid> <version>2.6.4</version> <scope>system</scope> <systempath>${basedir/hadoop/hadoop-yarn-common-2.6.4.jar</systempath> <artifactid>hadoop-yarn-server-common</artifactid> <version>2.6.4</version> <scope>system</scope> <systempath>${basedir/hadoop/hadoop-yarn-server-common-2.6.4.jar</systempath> <groupid>org.apache.hbase</groupid> <artifactid>hbase-client</artifactid> <version>1.1.3</version> <scope>system</scope> <systempath>${basedir/hadoop/hbase-client-1.1.3.jar</systempath> <groupid>org.apache.hbase</groupid> <artifactid>hbase-common</artifactid> <version>1.1.3</version> <scope>system</scope> <systempath>${basedir/hadoop/hbase-common-1.1.3.jar</systempath> 第 14 页共 41 页
<groupid>org.apache.hbase</groupid> <artifactid>hbase-protocol</artifactid> <version>1.1.3</version> <scope>system</scope> <systempath>${basedir/hadoop/hbase-protocol-1.1.3.jar</systempath> <groupid>org.apache.hbase</groupid> <artifactid>hbase-server</artifactid> <version>1.1.3</version> <scope>system</scope> <systempath>${basedir/hadoop/hbase-server-1.1.3.jar</systempath> </dependencies> 需要替换的 jar 包下载需要替换的 jar 包清单 : hadoop-annotations-2.5.1.jar hadoop-auth-2.5.1.jar hadoop-client-2.5.1.jar hadoop-common-2.5.1.jar hadoop-hdfs-2.5.1.jar hadoop-mapreduce-client-app-2.5.1.jar hadoop-mapreduce-client-common-2.5.1.jar hadoop-mapreduce-client-core-2.5.1.jar hadoop-mapreduce-client-jobclient-2.5.1.jar hadoop-mapreduce-client-shuffle-2.5.1.jar hadoop-yarn-api-2.5.1.jar hadoop-yarn-client-2.5.1.jar 第 15 页共 41 页
hadoop-yarn-common-2.5.1.jar hadoop-yarn-server-common-2.5.1.jar hbase-client-1.1.3.jar hbase-common-1.1.3.jar hbase-protocol-1.1.3.jar hbase-server-1.1.3.jar 以上 jar 包替换为 : hadoop-annotations-2.6.4.jar hadoop-auth-2.6.4.jar hadoop-client-2.6.4.jar hadoop-common-2.6.4.jar hadoop-hdfs-2.6.4.jar hadoop-mapreduce-client-app-2.6.4.jar hadoop-mapreduce-client-common-2.6.4.jar hadoop-mapreduce-client-core-2.6.4.jar hadoop-mapreduce-client-jobclient-2.6.4.jar hadoop-mapreduce-client-shuffle-2.6.4.jar hadoop-yarn-api-2.6.4.jar hadoop-yarn-client-2.6.4.jar hadoop-yarn-common-2.6.4.jar hadoop-yarn-server-common-2.6.4.jar hbase-client-1.1.3.jar hbase-common-1.1.3.jar hbase-protocol-1.1.3.jar hbase-server-1.1.3.jar 2). 如不使用 mvn 获取依赖, 您也可以在这里直接下载所有的客户端 jar 包 第 16 页共 41 页
Step3. 连接腾讯云参数设置 连接腾讯云参数设置 1). 连接腾讯云 Hbase 服务时必须设置以下参数为 true( 完整代码请参考示例代码 ), 方能正常使用 : config.setboolean("chbase.tencent.enable", true); 2). 如需要使用 yarn, 还需要额外设置实例 ID( 管理页面可以查到, 完整代码请参考示例代码 ), 如 : config.set("yarn.chbase.tencent.instanceid", "chb-lpvsvdlr"); 提示 : 除增加以上代码, 其余使用方式和社区版本 Hbase 一致 ; 可以参考 https://hbase.apache.org/ 的 API 文档 ; 如您未设置以上参数, 也可以正常连接社区版的 Hbase 第 17 页共 41 页
Step4. 操作示例代码 参考以下示例代码, 包括常见的 hbase 表格创建 删除, 插入 删除 读取数据等操作 import java.util.arraylist; import java.util.list; import java.util.map; import java.util.navigablemap; import org.apache.hadoop.conf.configuration; import org.apache.hadoop.hbase.cell; import org.apache.hadoop.hbase.cellutil; import org.apache.hadoop.hbase.hbaseconfiguration; import org.apache.hadoop.hbase.hcolumndescriptor; import org.apache.hadoop.hbase.htabledescriptor; import org.apache.hadoop.hbase.keyvalue; import org.apache.hadoop.hbase.tablename; import org.apache.hadoop.hbase.client.admin; import org.apache.hadoop.hbase.client.connection; import org.apache.hadoop.hbase.client.connectionfactory; import org.apache.hadoop.hbase.client.delete; import org.apache.hadoop.hbase.client.get; import org.apache.hadoop.hbase.client.put; import org.apache.hadoop.hbase.client.result; import org.apache.hadoop.hbase.client.resultscanner; import org.apache.hadoop.hbase.client.scan; import org.apache.hadoop.hbase.client.table; import org.apache.hadoop.hbase.util.bytes; public class Connect { public static void createtable (String tablename, String[] familys) throws Exception { Configuration config = HBaseConfiguration.create(); // 填写 zookeeper 地址, 多个地址用英文逗号隔开 第 18 页共 41 页
config.set("hbase.zookeeper.quorum", "10.66.133.178:2181"); // 设置重试参数 config.setint("hbase.client.retries.number", 1); /* * 要连接腾讯云的 hbase 服务必须设置此值为 true; 不设置该值功能和社区版相同, 可以正常连接自建 hbase 服务 */ config.setboolean("chbase.tencent.enable", true); TableName TABLE = TableName.valueOf(tableName); Connection con = ConnectionFactory.createConnection(config); Admin admin = con.getadmin(); if (admin.tableexists(table)) { System.out.println("table already exists!"); else { HTableDescriptor tabledesc = new HTableDescriptor(TABLE); for (int i = 0; i < familys.length; i++) { tabledesc.addfamily(new HColumnDescriptor(familys[i])); admin.createtable(tabledesc); System.out.println("create table " + tablename + " ok."); admin.close(); public static void deletetable(string tablename) throws Exception { Configuration config = HBaseConfiguration.create(); // 填写 zookeeper 地址, 多个地址用英文逗号隔开 config.set("hbase.zookeeper.quorum", "10.66.133.178:2181"); // 设置重试参数 config.setint("hbase.client.retries.number", 1); /* * 要连接腾讯云的 hbase 服务必须设置此值为 true; 不设置该值功能和社区版相同, 可以正常连接自建 hbase 服务 */ 第 19 页共 41 页
config.setboolean("chbase.tencent.enable", true); TableName TABLE = TableName.valueOf(tableName); Connection con = ConnectionFactory.createConnection(config); Admin admin = con.getadmin(); if (!admin.tableexists(table)) { System.out.println("table not exists!"); else { admin.disabletable(table); admin.deletetable(table); System.out.println("delete table " + tablename + " ok."); admin.close(); public static void get(string tablename, String rowkey) throws Exception { Configuration config = HBaseConfiguration.create(); // 填写 zookeeper 地址, 多个地址用英文逗号隔开 config.set("hbase.zookeeper.quorum", "10.66.133.178:2181"); // 设置重试参数 config.setint("hbase.client.retries.number", 1); /* * 要连接腾讯云的 hbase 服务必须设置此值为 true; 不设置该值功能和社区版相同, 可以正常连接自建 hbase 服务 */ config.setboolean("chbase.tencent.enable", true); Connection con = ConnectionFactory.createConnection(config); Table table = con.gettable(tablename.valueof(tablename)); Get get = new Get(rowkey.getBytes()); Result rs = table.get(get); for (Cell cell : rs.rawcells()) { System.out.print(new String(CellUtil.cloneRow(cell)) + " "); 第 20 页共 41 页
System.out.print(new String(CellUtil.cloneFamily(cell)) + ":"); System.out.print(new String(CellUtil.cloneQualifier(cell)) + " "); System.out.print(cell.getTimestamp() + " "); System.out.println(new String(CellUtil.cloneValue(cell))); table.close(); public static void del(string tablename, String rowkey) throws Exception { Configuration config = HBaseConfiguration.create(); // 填写 zookeeper 地址, 多个地址用英文逗号隔开 config.set("hbase.zookeeper.quorum", "10.66.133.178:2181"); // 设置重试参数 config.setint("hbase.client.retries.number", 1); /* * 要连接腾讯云的 hbase 服务必须设置此值为 true; 不设置该值功能和社区版相同, 可以正常连接自建 hbase 服务 */ config.setboolean("chbase.tencent.enable", true); Connection con = ConnectionFactory.createConnection(config); Table table = con.gettable(tablename.valueof(tablename)); // 批量删除 List<Delete> list = new ArrayList<Delete>(); Delete del = new Delete(rowkey.getBytes()); list.add(del); table.delete(list); // 单个删除 // Delete del = new Delete(Bytes.toBytes(rowkey)); // table.delete(del); System.out.println("del recored " + rowkey + " ok."); table.close(); public static void put 第 21 页共 41 页
(String tablename, String rowkey, String familyname, String colname, String valu e) throws Exception { Configuration config = HBaseConfiguration.create(); // 填写 zookeeper 地址, 多个地址用英文逗号隔开 config.set("hbase.zookeeper.quorum", "10.66.133.178:2181"); // 设置重试参数 config.setint("hbase.client.retries.number", 1); /* * 要连接腾讯云的 hbase 服务必须设置此值为 true; 不设置该值功能和社区版相同, 可以正常连接自建 hbase 服务 */ config.setboolean("chbase.tencent.enable", true); Connection con = ConnectionFactory.createConnection(config); Table table = con.gettable(tablename.valueof(tablename)); byte[] ROWKEY = Bytes.toBytes(rowkey); Put put = new Put(ROWKEY); Cell c1 = CellUtil.createCell(ROWKEY, Bytes.toBytes(familyname), Bytes.t obytes(colname), System.currentTimeMillis(), KeyValue.Type.Put.getCode(), Bytes.t obytes(value)); put.add(c1); table.put(put); System.out.println("--------------------put ok------------------"); table.close(); public static void scan (String tablename, String rowkey, String family) throws Exception { Configuration config = HBaseConfiguration.create(); // 填写 zookeeper 地址, 多个地址用英文逗号隔开 config.set("hbase.zookeeper.quorum", "10.66.133.178:2181"); 第 22 页共 41 页
// 设置重试参数 config.setint("hbase.client.retries.number", 1); /* * 要连接腾讯云的 hbase 服务必须设置此值为 true; 不设置该值功能和社区版相同, 可以正常连接自建 hbase 服务 */ config.setboolean("chbase.tencent.enable", true); Scan scan = new Scan(); scan.setstartrow(bytes.tobytes(rowkey)); scan.setcaching(500); scan.setcacheblocks(false); Connection con = ConnectionFactory.createConnection(config); Table table = con.gettable(tablename.valueof(tablename)); ResultScanner ss = table.getscanner(scan); System.out.println("--------------------------------------"); for (Result r : ss) { NavigableMap<byte[], byte []> map = r.getfamilymap(bytes.tobytes(family)); for (Map.Entry<byte[], byte[]> ent : map.entryset()) { String key = new String(ent.getKey()); String value = new String(ent.getValue()); System.out.println( "find result is:" + new String(r.getRow()) + " and code is:" + key + " and value is:" + value); System.out.println("--------------------------------------"); table.close(); public static void main(string[] args) throws Exception { if (args == null args.length < 1) { System.out.println("please input args..."); 第 23 页共 41 页
return; String op = args[0]; if (op.equals("create")) { // 要创建的列名 String[] familys = { "fam1", "fam2", "fam3" ; createtable("test", familys); else if (op.equals("put")) { put("test", "key1", "fam1", "col1", "value1"); else if (op.equals("get")) { get("test", "key1"); else if (op.equals("deltable")) { deletetable("test"); else if (op.equals("scan")) { scan("test", "key1", "fam1"); else if (op.equals("del")) { del("test", "key1"); 第 24 页共 41 页
MapReduce 示例代码 Map 类 import java.io.ioexception; import java.util.map; import java.util.navigablemap; import org.apache.hadoop.hbase.client.result; import org.apache.hadoop.hbase.io.immutablebyteswritable; import org.apache.hadoop.hbase.mapreduce.tablemapper; import org.apache.hadoop.hbase.util.bytes; import org.apache.hadoop.io.text; public class Mapper extends TableMapper<Text, Text>{ @Override protected void map (ImmutableBytesWritable rowkey, Result columns, Context context) throws IOException, InterruptedException{ NavigableMap<byte[], byte []> map = columns.getfamilymap(bytes.tobytes("retcode")); for (Map.Entry<byte[], byte[]> ent : map.entryset()) { String retcode=bytes.tostring(ent.getkey()); String value = Bytes.toString(ent.getValue()); Text retkey = new Text(retcode); Text retvalue = new Text(value); context.write(retkey, retvalue); Reduce 类 第 25 页共 41 页
import java.io.ioexception; import org.apache.hadoop.hbase.io.immutablebyteswritable; import org.apache.hadoop.hbase.mapreduce.tablereducer; import org.apache.hadoop.io.text; public class Reduce extends TableReducer<Text, Text, ImmutableBytesWritable> { @Override protected void cleanup(context context) throws IOException, InterruptedException { @Override protected void setup(context context) throws IOException, InterruptedException { super.setup(context); @Override public void reduce (Text key, Iterable<Text> valu es, Context context) throws IOException, InterruptedException { long size = 0; String rowkey = key.tostring(); System.out.println(rowkey); for(text t:values){ System.out.println(t.toString()); 第 26 页共 41 页
提交任务 import org.apache.hadoop.conf.configuration; import org.apache.hadoop.hbase.client.scan; import org.apache.hadoop.hbase.mapreduce.tablemapreduceutil; import org.apache.hadoop.hbase.util.bytes; import org.apache.hadoop.io.text; import org.apache.hadoop.mapreduce.job; import org.apache.hadoop.mapreduce.lib.output.nulloutputformat; public class MR { public static void main(string[] args) throws Exception{ Job job =Job.getInstance(); Configuration conf=job.getconfiguration(); // 填写 zookeeper 地址, 多个地址用英文逗号隔开 conf.set("hbase.zookeeper.quorum", "10.66.133.178:2181"); // 必填 : 填写腾讯云 Hbase 实例 ID conf.set("yarn.chbase.tencent.instanceid", "chb-lpvsvdlr"); job.setjobname("testjob"); String tablename = "monitordata_201603"; Scan scan = new Scan(); scan.setstartrow(bytes.tobytes("0800:00_00000000")); scan.setcaching(500); scan.setcacheblocks(false); job.setjarbyclass(mr.class); job.setreducerclass(reduce.class); job.setoutputformatclass(nulloutputformat.class); job.setnumreducetasks(5); // 如果要使用第三方 jar 包, 可使用该工具类上传 //job.addfiletoclasspath(jobhelper.addjartodistributedcache(genericobjectpoolcon fig.class, conf)); 第 27 页共 41 页
TableMapReduceUtil.initTableMapperJob(tableName, scan, Mapper.class, Tex t.class, Text.class, job); boolean b = job.waitforcompletion(true); if (!b) { throw new Exception("error with job!"); 工具类 ( 如果需要使用第三方 jar 包 ) import java.io.ioexception; import org.apache.hadoop.fs.path; import org.apache.hadoop.fs.filesystem; import org.apache.hadoop.conf.configuration; import java.io.file; public class JobHelper { public JobHelper() { public static Path addjartodistributedcache (Class classtoadd, Configuration conf) throws IOException { n().getpath(); // Retrieve jar file for class2add String jar = classtoadd.getprotectiondomain().getcodesource().getlocatio File jarfile = new File(jar); 第 28 页共 41 页
// Declare new HDFS location Path hdfsjar = new Path("/tmp/hadoop/userlib/" + jarfile.getname()); // Mount HDFS FileSystem hdfs = FileSystem.get(conf); // Copy (override) jar file to HDFS hdfs.copyfromlocalfile(false, true, new Path(jar), hdfsjar); hdfs.close(); return hdfsjar; 第 29 页共 41 页
endpoint 协处理器示例代码 使用协处理, 需要将 jar 包提供给我们, 我们安装后提供 hdfs 地址, 您可以再通过 shell 或者 api 方式 ( 代码中有示例 ) 安装 强烈建议开发好协处理器 jar 包后, 经过充分测试再安装, 避免安装后影响 Hbase 服务的正常使用 协处理器开发步骤 1. 编写 proto 文件 ( 示例 :RowCount.proto); 因为 hbase 使用 google 的 protoc-2.5.0 版本, 所以最好使用相同版本编译 proto 文件, 生成 java 文件 ( 示例 :RowCountService.java); 2. 编写服务端 EndPoint 代码 ( 示例 :RowCountEndPoint.java); 3. 编写客户端调用 Client 代码 ( 示例 :RowCountClient.java); 4. 编译 jar 包 endpoint 协处理器示例 : proto 文件 : option java_package = "com.tencent.yun.endpoint.proto"; option java_outer_classname = "RowCountService"; option java_generic_services = true; option java_generate_equals_and_hash = true; option optimize_for = SPEED; message RowCountRequest{ required string family = 1; required string column = 2; message RowCountResponse { required int64 rowcount = 1 [default = 0]; service RowCount { 第 30 页共 41 页
rpc getrowcount(rowcountrequest) returns (RowCountResponse); 服务端 RowCountEndPoint 类 : import java.io.ioexception; import java.util.arraylist; import java.util.list; import org.apache.hadoop.hbase.cell; import org.apache.hadoop.hbase.coprocessor; import org.apache.hadoop.hbase.coprocessorenvironment; import org.apache.hadoop.hbase.client.scan; import org.apache.hadoop.hbase.coprocessor.coprocessorexception; import org.apache.hadoop.hbase.coprocessor.coprocessorservice; import org.apache.hadoop.hbase.coprocessor.regioncoprocessorenvironment; import org.apache.hadoop.hbase.protobuf.responseconverter; import org.apache.hadoop.hbase.regionserver.internalscanner; import org.apache.hadoop.hbase.util.bytes; import com.google.protobuf.rpccallback; import com.google.protobuf.rpccontroller; import com.google.protobuf.service; import com.tencent.yun.endpoint.proto.rowcountservice; import com.tencent.yun.endpoint.proto.rowcountservice.rowcountrequest; import com.tencent.yun.endpoint.proto.rowcountservice.rowcountresponse; public class RowCountEndPoint extends RowCountService.RowCount implements Coprocessor, CoprocessorService { private RegionCoprocessorEnvironment env; 第 31 页共 41 页
public Service getservice() { return this; 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 region!"); public void stop(coprocessorenvironment arg0) throws IOException { // do nothing @Override public void getrowcount (RpcController controller, RowCountRequest request, RpcCallback<RowCountResponse > done) { Scan scan = new Scan(); scan.addfamily(bytes.tobytes(request.getfamily())); scan.addcolumn(bytes.tobytes(request.getfamily()), Bytes.toBytes(request.getColumn())); // scan.setmaxversions(1); InternalScanner scanner = null; RowCountResponse response = null; long count = 0L; try { List<Cell> results = new ArrayList<Cell>(); boolean hasmore = false; scanner = env.getregion().getscanner(scan); 第 32 页共 41 页
do { hasmore = scanner.next(results); for (Cell cell : results) { count++; // count = count + Bytes.toLong(CellUtil.cloneValue(cell)); results.clear(); while (hasmore); response = RowCountResponse.newBuilder().setRowCount(count).build(); catch (IOException e) { ResponseConverter.setControllerException(controller, e); finally { if (scanner!= null) { try { scanner.close(); catch (IOException ignored) { done.run(response); 客户端 RowCountClient 类 : import java.io.ioexception; import java.util.map; import org.apache.hadoop.conf.configuration; 第 33 页共 41 页
import org.apache.hadoop.fs.path; import org.apache.hadoop.hbase.coprocessor; import org.apache.hadoop.hbase.hbaseconfiguration; import org.apache.hadoop.hbase.hcolumndescriptor; import org.apache.hadoop.hbase.htabledescriptor; import org.apache.hadoop.hbase.tablename; import org.apache.hadoop.hbase.client.admin; import org.apache.hadoop.hbase.client.connection; import org.apache.hadoop.hbase.client.connectionfactory; import org.apache.hadoop.hbase.client.table; import org.apache.hadoop.hbase.client.coprocessor.batch; import org.apache.hadoop.hbase.ipc.blockingrpccallback; import org.apache.hadoop.hbase.ipc.serverrpccontroller; import com.google.protobuf.serviceexception; import com.tencent.yun.endpoint.proto.rowcountservice.rowcount; import com.tencent.yun.endpoint.proto.rowcountservice.rowcountrequest; import com.tencent.yun.endpoint.proto.rowcountservice.rowcountresponse; import com.tencent.yun.observer.regionobserver; public class RowCountClient { public static void testrowcountendpoint(string tablename) { System.out.println("begin test..."); Configuration config = HBaseConfiguration.create(); // 填写 hbase zk 地址 config.set("hbase.zookeeper.quorum", "100.67.159.134:2181,100.67.159.141:2181,100.67.159.196:2181"); // 填寫 family 名和列名 final RowCountRequest r eq = RowCountRequest.newBuilder().se tfamily("family").setcolumn("field0").build(); 第 34 页共 41 页
RowCountResponse resp = null; try { Connection con = ConnectionFactory.createConnection(config); Table table = con.gettable(tablename.valueof(tablename)); Map<byte [], Long> results = table.coprocessorservice(rowcount.class, null, null, new Batch.Call<RowCount, Long>() { public Long call(rowcount instance) throws IOException { ServerRpcController controller = new ServerRpcController(); BlockingRpcCallback<RowCountResp onse> rpccall = new BlockingRpcCallback<RowCountResponse>(); instance.getrowcount(controller, req, rpccall); RowCountResponse resp = rpccall.get(); // return resp.hasrowcount()? resp.getrowcount() : 0L; ); long count = 0L; for (Long sum : results.values()) { System.out.println("region row Sum = " + sum); count += sum; System.out.println("total count = " + count); catch (IOException e) { e.printstacktrace(); catch (ServiceException e) { e.printstacktrace(); catch (Throwable e) { 第 35 页共 41 页
e.printstacktrace(); finally{ public static void delcorprocessor(string tablename) { System.out.println("begin delcorprocessor..."); try { Configuration config = HBaseConfiguration.create(); // 填写 hbase zk 地址 config.set("hbase.zookeeper.quorum", "100.67.159.134:2181,100.67.159.141:2181,100.67.159.196:2181"); TableName TABLE = TableName.valueOf(tableName); Connection con = ConnectionFactory.createConnection(config); Admin admin = con.getadmin(); HTableDescriptor tabledesc = admin.gettabledescriptor(table); tabledesc.removecoprocessor(rowcountendpoint.class.getcanonicalname( )); tabledesc.removecoprocessor(regionobserver.class.getcanonicalname()) ; admin.modifytable(table, tabledesc); catch (IOException e) { e.printstacktrace(); System.out.println("end delcorprocessor...ok"); public static void setuptoexisttable(string tablename) { 第 36 页共 41 页
System.out.println("begin setuptoexisttable..."); try { Configuration config = HBaseConfiguration.create(); // 填写 hbase zk 地址 config.set("hbase.zookeeper.quorum", "100.67.159.134:2181,100.67.159.141:2181,100.67.159.196:2181"); TableName TABLE = TableName.valueOf(tableName); Connection con = ConnectionFactory.createConnection(config); Admin admin = con.getadmin(); HTableDescriptor tabledesc = admin.gettabledescriptor(table); // 填写我们提供的 jar 包的 hdfs 地址 Path jarpath = new Path( "hdfs://100.67.159.132:8020/coprocessor/1/thbase-1.0-snapshot.jar"); tabledesc.addcoprocessor(rowcountendpoint.class.getcanonicalname(), jarpath, Coprocessor.PRIORITY_USER, null); tabledesc.addcoprocessor(regionobserver.class.getcanonicalname(), ja rpath, Coprocessor.PRIORITY_USER, null); admin.modifytable(table, tabledesc); catch (IOException e) { e.printstacktrace(); System.out.println("end setuptoexisttable...ok"); public static void createandsetup(string tablename) { System.out.println("begin safesetup..."); try { 第 37 页共 41 页
Configuration config = HBaseConfiguration.create(); // 填写 hbase zk 地址 config.set("hbase.zookeeper.quorum", "100.67.159.134:2181,100.67.159.141:2181,100.67.159.196:2181"); TableName TABLE = TableName.valueOf(tableName); Connection con = ConnectionFactory.createConnection(config); Admin admin = con.getadmin(); HTableDescriptor tabledesc = new HTableDescriptor(TABLE); HColumnDescriptor columnfamily1 = new HColumnDescriptor("f1"); columnfamily1.setmaxversions(3); tabledesc.addfamily(columnfamily1); // 填写我们提供的 jar 包的 hdfs 地址 Path jarpath = new Path( "hdfs://100.67.159.132:8020/coprocessor/1/thbase-1.0-snapshot.jar"); tabledesc.addcoprocessor(rowcountendpoint.class.getcanonicalname(), jarpath, Coprocessor.PRIORITY_USER, null); tabledesc.addcoprocessor(regionobserver.class.getcanonicalname(), ja rpath, Coprocessor.PRIORITY_USER, null); admin.createtable(tabledesc); System.out.println("end safesetup...ok"); catch (IOException e) { e.printstacktrace(); public static void main(string[] args) { 第 38 页共 41 页
if (args == null args.length < 2) { System.out.println("please input args..."); return; String op = args[0]; String tablename = args[1]; if (op.equals("setup")) { setuptoexisttable(tablename); else if (op.equals("safesetup")) { createandsetup(tablename); else if (op.equals("run")) { testrowcountendpoint(tablename); else if (op.equals("unset")) { delcorprocessor(tablename); 第 39 页共 41 页
observer 协处理器示例 observer 协处理器示例 : import java.io.ioexception; import java.util.list; import org.apache.hadoop.hbase.cell; import org.apache.hadoop.hbase.cellutil; import org.apache.hadoop.hbase.client.get; import org.apache.hadoop.hbase.coprocessor.baseregionobserver; import org.apache.hadoop.hbase.coprocessor.observercontext; import org.apache.hadoop.hbase.coprocessor.regioncoprocessorenvironment; import org.apache.hadoop.hbase.util.bytes; public class RegionObserver extends BaseRegionObserver { private static final byte[] ADMIN = Bytes.toBytes("admin"); private static final byte[] COLUMN_FAMILY = Bytes.toBytes("f1"); private static final byte[] COLUMN = Bytes.toBytes("col1"); private static final byte[] VALUE = Bytes.toBytes( "You can not see Admin details"); @Override public void pregetop (O bser verconte xt<regioncoproce ssorenvironment> e, Get get, List<Cell> results) throws IOException{ if (Bytes.equals(get.getRow(),ADMIN)) { Cell c = CellUtil.createCell(get.getRow(), COLUMN_FAMILY, COLUMN, Sy stem.currenttimemillis(), (byte)4, VALUE); results.add(c); e.bypass(); 第 40 页共 41 页
Powered by TCPDF (www.tcpdf.org) 操作指南产品文档 第 41 页共 41 页