- 浏览: 527136 次
- 性别:
- 来自: 杭州
文章分类
最新评论
-
飞天奔月:
public List<String> gener ...
实践中的重构30_不做油漆匠 -
在世界的中心呼喚愛:
在世界的中心呼喚愛 写道public class A {
...
深入理解ReferenceQueue GC finalize Reference -
在世界的中心呼喚愛:
在世界的中心呼喚愛 写道在世界的中心呼喚愛 写道在classB ...
深入理解ReferenceQueue GC finalize Reference -
在世界的中心呼喚愛:
在世界的中心呼喚愛 写道在classB的finalize上打断 ...
深入理解ReferenceQueue GC finalize Reference -
在世界的中心呼喚愛:
iteye比较少上,如果可以的话,可以发e-mail交流:ch ...
深入理解ReferenceQueue GC finalize Reference
hbase的CoprocessorProtocol及一个简单的通用扩展实现V2
http://zhang-xzhi-xjtu.iteye.com/blog/2163321
hbase中的CoprocessorProtocol机制.
CoprocessorProtocol的原理比较简单,近似于一个mapreduce框架。由client将scan分解为面向多个region的请求,并行发送请求到多个region,然后client做一个reduce的操作,得到最后的结果。
先看一个例子,使用hbase的AggregationClient可以做到简单的面向单个column的统计。
看下hbase的源码。AggregateImplementation
这里由于
所以,hbase自带的Aggregate函数,只能面向单列进行统计。
当我们想对多列进行Aggregate,并同时进行countRow时,有以下选择。
1 scan出所有的row,程序自己进行Aggregate和count。
2 使用AggregationClient,调用多次,得到所有的结果。由于多次调用,有一致性问题。
3 自己扩展CoprocessorProtocol。
首先我们可以写一个protocol的通用框架。
定义protocol接口。
定义该protocol的实现。
定义一个rowHandler。
定义一个reduce。
定义一个client。
这样,我们就有了一个protocol的通用框架。
假设我们要一个count的功能。
则只需要实现对应的handler和reducer。
假设我们要实现多个列的sum和全部结果的row,我们也只是通过添加hander,reducer和result来实现。
有了CoprocessorProtocol,可以扩展出来很多的功能,这个机制还是很强大的。
http://zhang-xzhi-xjtu.iteye.com/blog/2163321
hbase中的CoprocessorProtocol机制.
CoprocessorProtocol的原理比较简单,近似于一个mapreduce框架。由client将scan分解为面向多个region的请求,并行发送请求到多个region,然后client做一个reduce的操作,得到最后的结果。
先看一个例子,使用hbase的AggregationClient可以做到简单的面向单个column的统计。
@Test public void testAggregationClient() throws Throwable { LongColumnInterpreter columnInterpreter = new LongColumnInterpreter(); AggregationClient aggregationClient = new AggregationClient( CommonConfig.getConfiguration()); Scan scan = new Scan(); scan.addColumn(ColumnFamilyName, QName1); Long max = aggregationClient.max(TableNameBytes, columnInterpreter, scan); Assert.assertTrue(max.longValue() == 100); Long min = aggregationClient.min(TableNameBytes, columnInterpreter, scan); Assert.assertTrue(min.longValue() == 20); Long sum = aggregationClient.sum(TableNameBytes, columnInterpreter, scan); Assert.assertTrue(sum.longValue() == 120); Long count = aggregationClient.rowCount(TableNameBytes, columnInterpreter, scan); Assert.assertTrue(count.longValue() == 4); }
看下hbase的源码。AggregateImplementation
@Override public <T, S> T getMax(ColumnInterpreter<T, S> ci, Scan scan) throws IOException { T temp; T max = null; InternalScanner scanner = ((RegionCoprocessorEnvironment) getEnvironment()) .getRegion().getScanner(scan); List<KeyValue> results = new ArrayList<KeyValue>(); byte[] colFamily = scan.getFamilies()[0]; byte[] qualifier = scan.getFamilyMap().get(colFamily).pollFirst(); // qualifier can be null. try { boolean hasMoreRows = false; do { hasMoreRows = scanner.next(results); for (KeyValue kv : results) { temp = ci.getValue(colFamily, qualifier, kv); max = (max == null || (temp != null && ci.compare(temp, max) > 0)) ? temp : max; } results.clear(); } while (hasMoreRows); } finally { scanner.close(); } log.info("Maximum from this region is " + ((RegionCoprocessorEnvironment) getEnvironment()).getRegion() .getRegionNameAsString() + ": " + max); return max; }
这里由于
byte[] colFamily = scan.getFamilies()[0]; byte[] qualifier = scan.getFamilyMap().get(colFamily).pollFirst();
所以,hbase自带的Aggregate函数,只能面向单列进行统计。
当我们想对多列进行Aggregate,并同时进行countRow时,有以下选择。
1 scan出所有的row,程序自己进行Aggregate和count。
2 使用AggregationClient,调用多次,得到所有的结果。由于多次调用,有一致性问题。
3 自己扩展CoprocessorProtocol。
首先我们可以写一个protocol的通用框架。
定义protocol接口。
public interface MyCoprocessorProtocol extends CoprocessorProtocol { public static final long VERSION = 1L; public <T> T handle(RowHandler<T> rowHandler, Scan scan) throws IOException; }
定义该protocol的实现。
public class MyEndpointImpl extends BaseEndpointCoprocessor implements MyCoprocessorProtocol { protected static Log log = LogFactory.getLog(MyEndpointImpl.class); @Override public ProtocolSignature getProtocolSignature(String protocol, long version, int clientMethodsHashCode) throws IOException { if (MyCoprocessorProtocol.class.getName().equals(protocol)) { return new ProtocolSignature(MyCoprocessorProtocol.VERSION, null); } throw new IOException("Unknown protocol: " + protocol); } @Override public <T> T handle(RowHandler<T> rowHandler, Scan scan) throws IOException { InternalScanner scanner = ((RegionCoprocessorEnvironment) getEnvironment()) .getRegion().getScanner(scan); List<KeyValue> results = new ArrayList<KeyValue>(); T t = rowHandler.getInitValue(); try { boolean hasMoreRows = false; do { hasMoreRows = scanner.next(results); log.debug("scanner result : " + results + " hasMoreRows = " + hasMoreRows); t = rowHandler.handle(results, t); results.clear(); } while (hasMoreRows); } finally { scanner.close(); } return t; } }
定义一个rowHandler。
public interface RowHandler<T> extends Writable { public T getInitValue(); public T handle(List<KeyValue> keyValues, T t); }
定义一个reduce。
public interface MyReducer<T, R> { public R getInitValue(); public R reduce(R r, T t); }
定义一个client。
public class MyClient { HTableInterface table; public MyClient(HTableInterface table) { this.table = table; } public <T, R> R call(final byte[] tableName, final RowHandler<T> howHandler, final MyReducer<T, R> myReducer, final Scan scan) throws Throwable { class MyCallBack implements Batch.Callback<T> { R r = myReducer.getInitValue(); R getResult() { return r; } @Override public synchronized void update(byte[] region, byte[] row, T result) { r = myReducer.reduce(r, result); } } MyCallBack myCallBack = new MyCallBack(); try { table.coprocessorExec(MyCoprocessorProtocol.class, scan.getStartRow(), scan.getStopRow(), new Batch.Call<MyCoprocessorProtocol, T>() { @Override public T call(MyCoprocessorProtocol instance) throws IOException { return instance.handle(howHandler, scan); } }, myCallBack); } finally { table.close(); } return myCallBack.getResult(); } }
这样,我们就有了一个protocol的通用框架。
假设我们要一个count的功能。
则只需要实现对应的handler和reducer。
public class CountHandler implements RowHandler<Long> { @Override public void readFields(DataInput arg0) throws IOException { } @Override public void write(DataOutput arg0) throws IOException { } @Override public Long getInitValue() { return 0L; } @Override public Long handle(List<KeyValue> keyValues, Long t) { if (!keyValues.isEmpty()) { return t + 1; } else { return t; } } } public class CountReducer implements MyReducer<Long, Long> { @Override public Long getInitValue() { return 0L; } @Override public Long reduce(Long r, Long t) { return r + t; } }
假设我们要实现多个列的sum和全部结果的row,我们也只是通过添加hander,reducer和result来实现。
public class CountAndSumResult implements Writable { private List<Long> resultList = new ArrayList<Long>(); private Long count = 0L; public CountAndSumResult() { } public CountAndSumResult(int resultSize) { for (int i = 0; i < resultSize; i++) { resultList.add(0L); } } public Long getCount() { return count; } public void setCount(Long count) { this.count = count; } public Long getSum(int i) { return resultList.get(i); } public void setSum(int i, Long sum) { resultList.set(i, sum); } public int getResultSize() { return resultList.size(); } @Override public void write(DataOutput out) throws IOException { out.writeLong(count); out.writeInt(resultList.size()); for (Long v : resultList) { out.writeLong(v); } } @Override public void readFields(DataInput in) throws IOException { count = in.readLong(); int size = in.readInt(); for (int i = 0; i < size; i++) { resultList.add(in.readLong()); } } } public class CountAndSumHandler implements RowHandler<CountAndSumResult> { private List<String> columns = new ArrayList<String>(); public CountAndSumHandler() { } public CountAndSumHandler(List<String> columns) { super(); this.columns = columns; } @Override public void write(DataOutput out) throws IOException { out.writeInt(columns.size()); for (String s : columns) { out.writeUTF(s); } } @Override public void readFields(DataInput in) throws IOException { int size = in.readInt(); for (int i = 0; i < size; i++) { columns.add(in.readUTF()); } } @Override public CountAndSumResult handle(List<KeyValue> keyValues, CountAndSumResult t) { if (!keyValues.isEmpty()) { t.setCount(t.getCount() + 1); } for (int i = 0; i < columns.size(); i++) { String column = columns.get(i); for (KeyValue kv : keyValues) { if (column.equals(Bytes.toString(kv.getQualifier()))) { byte[] value = kv.getValue(); if (value == null || value.length == 0) { } else { Long tValue = Bytes.toLong(value); t.setSum(i, t.getSum(i) + tValue); } break; } } } return t; } @Override public CountAndSumResult getInitValue() { return new CountAndSumResult(columns.size()); } } public class CountAndSumReducer implements MyReducer<CountAndSumResult, CountAndSumResult> { @Override public CountAndSumResult getInitValue() { return null; } @Override public CountAndSumResult reduce(CountAndSumResult r, CountAndSumResult t) { if (r == null) { return t; } if (t == null) { return r; } r.setCount(r.getCount() + t.getCount()); int size = r.getResultSize(); for (int i = 0; i < size; i++) { r.setSum(i, r.getSum(i) + t.getSum(i)); } return r; } }
有了CoprocessorProtocol,可以扩展出来很多的功能,这个机制还是很强大的。
评论
4 楼
zhang_xzhi_xjtu
2014-12-13
3 楼
zhang_xzhi_xjtu
2014-12-04
关于兼容性,可以看
hbase 0.94.0 0.94.9 0.94.24 功能不兼容初步分析
http://zhang-xzhi-xjtu.iteye.com/blog/2163258
通用的代码实现
hbase的CoprocessorProtocol及一个简单的通用扩展实现V2
http://zhang-xzhi-xjtu.iteye.com/blog/2163321
代码地址
https://github.com/zhang-xzhi/simplehbase
hbase 0.94.0 0.94.9 0.94.24 功能不兼容初步分析
http://zhang-xzhi-xjtu.iteye.com/blog/2163258
通用的代码实现
hbase的CoprocessorProtocol及一个简单的通用扩展实现V2
http://zhang-xzhi-xjtu.iteye.com/blog/2163321
代码地址
https://github.com/zhang-xzhi/simplehbase
2 楼
zhang_xzhi_xjtu
2014-11-19
暂时没有发现问题,不过建议你自己理解后,再使用。
另外 我这边有一个hbase的orm框架,可以看看有什么建议。
https://github.com/zhang-xzhi/simplehbase
http://zhang-xzhi-xjtu.iteye.com/blog/2056369
另外 我这边有一个hbase的orm框架,可以看看有什么建议。
https://github.com/zhang-xzhi/simplehbase
http://zhang-xzhi-xjtu.iteye.com/blog/2056369
1 楼
liu12qw
2014-11-19
您好,我这边hbase不能重启太多次测试,请问只要打成lib放到各个节点就能直接用吗,后面有没有发现这个代码的其他问题?
发表评论
-
hbase分页功能的几种实现方案
2015-01-13 23:52 5383hbase分页功能的几种实现方案。 分页功能是线上系统的常用 ... -
simplehbase v0.98.1开始支持hbase0.98
2014-12-29 21:48 1025https://github.com/zhang-xzhi/s ... -
hbase轻量级中间件simplehbase v1.0简介
2014-12-13 18:55 1374https://github.com/zhang-xzhi/s ... -
hbase put UML图
2014-12-11 23:40 1277create Htable put hbase rpc ... -
hbase开发问题-PooledHTable多次close导致问题
2014-12-11 23:27 2006PooledHTable多次close导致问题 Pooled ... -
hbase的CoprocessorProtocol及一个简单的通用扩展实现V2
2014-12-04 18:00 2081hbase中的CoprocessorProtocol机制. ... -
hbase 0.94.0 0.94.9 0.94.24 功能不兼容初步分析
2014-12-04 16:10 1093hbase 0.94.0 0.94.9 0.94.24 功能不 ... -
simplehbase对JOPO新增xml配置和无配置方式
2014-10-24 22:50 979simplehbase介绍文章如下: https://gith ... -
hbase轻量级中间件simplehbase v0.9简介
2014-07-14 13:57 621https://github.com/zhang-xzhi/s ... -
hbase轻量级中间件simplehbase v0.8简介
2014-04-28 21:44 3739https://github.com/zhang-xzhi/s ... -
hbase开发问题-hbase版本号报错
2014-04-22 19:19 2748由于使用了自定义的classloader,导致报错。 p ... -
HBase Client使用注意点
2014-04-21 12:51 2532HBase Client使用注意点: 1 HTable线程 ... -
hbase开发问题-hbase-0.94.0的ServerCallable callTimeout处理有问题
2014-04-14 22:07 2172读hbase-0.94.0的ServerCallable时,发 ... -
Phoenix和simplehbase功能简单比较
2014-04-02 17:20 1613Phoenix和simplehbase功能简单比较 大数据应 ... -
hbase web console simplehbaseviewer
2014-03-12 19:11 1226https://github.com/zhang-xzhi/s ... -
hbase轻量级中间件simplehbase v0.2简介
2013-12-19 23:51 1666https://github.com/zhang-xzhi/s ... -
hbase轻量级中间件simplehbase v0.1简介
2013-10-09 19:29 1538simplehbase尝试简化基于hbase的java应用开发 ... -
hbase的基本操作
2013-08-18 14:02 4463本文列举一些hbase的基本操作代码。 package ... -
hadoop_hadoop的一次读取
2013-04-29 13:09 1861一次hadoop的read getFileSystem 代码 ... -
hadoop_hadoop的map reduce
2011-11-09 21:21 1229这个根据功能模块分为 ...
相关推荐
首先,我们需要创建一个HBase的配置对象,指定ZooKeeper的地址和端口号。 ```java private static Configuration config = null; private static HTablePool tp = null; static { config = HBaseConfiguration....
HBase原理及实例
Hbase是Apache的NoSQL分布式可扩展Hadoop数据库,可以很好地横向扩展。Hbase中的数据是面向列的数据库,其中结构化数据存储在键值对中。Hbase用Java编写。Hbase的灵感来自Google Paper-“大表:结构化数据的分布式...
hbase分页查询实现.pdf
HBase_SI_--_实现HBase_ACID的理论
节制2013年12月23日最新的hadoop和hbase兼容版本搭建 hadoop-2.2.0 hbase-0.96.1.1 java Hbase java DBHelper CRUD等通用方法 花了两天时间整理的,含有lib包 & 源码
实现了HBase的存取,很有借鉴性。是一个在eclipse运行的完整代码
删除Hbase中某个表的一列值 命令 java -jar deleteOneColumn.jar(这个文件的路径) '表名' '列簇名' '列名'
java 利用 sping-data-hadoop HbaseTemplate 操作hbase find get execute 等方法 可以直接运行
HBase 是一个开源的、分布式的、版本化的 NoSQL 数据库(也即非关系型数据库),它利用 Hadoop 分布式文件系统(Hadoop Distributed File System,HDFS)提供分布式数据存储。与传统的关系型数据库类似,HBase 也以...
HBase(hbase-2.4.9-bin.tar.gz)是一个分布式的、面向列的开源数据库,该技术来源于 Fay Chang 所撰写的Google论文“Bigtable:一个结构化数据的分布式存储系统”。就像Bigtable利用了Google文件系统(File System...
在本次实验的第一题是用编程Java API实现指定功能,并用Hadoop提供的HBase Shell命令完成相同任务。在本题中学习了列出HBase所有的表的相关信息,在终端打印出指定的表的所有记录数据,向已经创建好的表添加和删除...
hbase在小米的应用现状、改进与扩展 应用场景:云服务、推送
来自一位老学姐的Hbase安装详细教程(清华大学镜像下)及基本操作,希望能带给你们些许帮助。主要包含Hbase的下载过程及其遇到的小问题,后续会继续完善该文档!
这是hbase对数据存储的代码实现,让你轻松秒懂hbase,
python3 使用 thrift 操作hbase 安装hbase-thirft后有一个Hbase报错 使用这个修改完成的Hbase类替换掉原来的Hbase类问题全部解决 主要是因为python版本兼容性带来的问题
搭建pinpoint需要的hbase初始化脚本hbase-create.hbase
3. 如果设计一个笔记的表,表中要求有笔记的属性和笔记的内容,怎么做 4. HBase部署时如何指定多个zookeeper 5. HBase shell是基于哪种JVM运行的语言实现的 6. HBase shell如何写过滤条件 7. 如何删除HBase中的表 8....
对HBase的API做了一层抽象,统一了HBase1.x和HBase2.x的实现,并提供了读写HBase的ORM的支持,同时,sdk还对HBase thrift 的客户端API进行了池化封装,(类似JedisPool),消除了直接使用原生API的各种问题,使之...
基于hadoop+hbase+springboot实现的分布式网盘系统,适合本科毕业设计 资源包含的整个demo在Hadoop,和Hbase环境搭建好了,可以启动起来。 技术选型 1.Hadoop 2.Hbase 3.SpringBoot ...... 系统实现的功能 1.用户...