读书笔记-HBase in Action-第二部分Advanced concepts-(2)Coprocessor

读书笔记-HBase in Action-第二部分Advanced concepts-(2)Coprocessor

大家好,又见面了,我是全栈君。

Coprocessor是HBase 0.92.0引入的特性。使用Coprocessor。能够将一些计算逻辑下推到HBase节点,HBase由一个单纯的存储系统升级为分布式数据处理平台。

Coprocessor分为两种:Observer和Endpoint。

Observer能改动扩展已有的client操作功能。而Endpoint能引入新的client操作。

Observer

Observer的作用类似于数据库的触发器或者AOP中的advice。下图为Put操作添加Observer,当中1-2-4-6是一次正常的Put操作RPC调用过程,而3和5属于Observer,能够在Put操作之前和之后添加自己定义处理逻辑。

读书笔记-HBase in Action-第二部分Advanced concepts-(2)Coprocessor

Observer包含三种,RegionObserver(针对数据訪问和更新操作,执行在Region上)/WALObserver(针对WAL日志事件,执行在RegionServer上下文)/MasterObserver(针对DDL操作,执行在Master节点)。

Endpoint

Endpoint的作用则类似于数据库存储过程。实现机制是通过扩展HBase RPC协议,给client暴露新的操作接口。

例如以下图,client负责发起调用和收集结果,服务端各节点负责并行计算。

读书笔记-HBase in Action-第二部分Advanced concepts-(2)Coprocessor

实战

以上一章的follows表为例,通过Observer实现followedBy被关注表数据一致性维护。Endpoint实现关注人数量统计。

由于要实如今插入follows表时自己主动插入followedBy表。须要用到关注人/被关注人username信息,所以首先升级schema。

读书笔记-HBase in Action-第二部分Advanced concepts-(2)Coprocessor

实现Observer

代码中有三处凝视值得注意:

  1. postPut方法在put操作之后被调用。
  2. 假设通过hbase-site.xml安装Observer。会应用到全局全部表,所以这里推断put操作的是否follows表。

  3. 这里有点bad smell。Observer执行在服务器端。为了共用代码,又调用client代码,仅为演示作用。

packageHBaseIA.TwitBase.coprocessors;//…publicclass FollowsObserver extends BaseRegionObserver {    private HTablePool pool = null;    @Override    public void start(CoprocessorEnvironment env)throws IOException {        pool = newHTablePool(env.getConfiguration(), Integer.MAX_VALUE);    }    @Override    public void stop(CoprocessorEnvironment env)throws IOException {        pool.close();    }    @Override    public void postPut(//1,在Put操作之后调用            finalObserverContext<RegionCoprocessorEnvironment> e,            final Put put,            final WALEdit edit,            final boolean writeToWAL) throws IOException {        byte[] table=e.getEnvironment().getRegion().getRegionInfo().getTableName();        if (!Bytes.equals(table,FOLLOWS_TABLE_NAME))             return;  //2,推断表名        KeyValue kv =put.get(RELATION_FAM, FROM).get(0);        String from =Bytes.toString(kv.getValue());        kv = put.get(RELATION_FAM,TO).get(0);        String to =Bytes.toString(kv.getValue());        RelationsDAO relations = newRelationsDAO(pool);        relations.addFollowedBy(to,from);//3,插入followedBy表    }}

Observer的安装能够通过改动hbase-site.xml或者使用tableschema改动语句完毕,前者须要重新启动HBase服务,后者仅仅须要又一次上下线相应表。

$ hbase shellHBaseShell; enter 'help<RETURN>' for list of supported commands.Type"exit<RETURN>" to leave the HBase ShellVersion0.92.0, r1231986, Mon Jan 16 13:16:35 UTC 2012hbase(main):001:0>disable 'follows'0 row(s) in 7.0560 secondshbase(main):002:0>alter 'follows', METHOD => 'table_att','coprocessor'=>'file:///Users/ndimiduk/repos/hbaseiatwitbase/target/twitbase-1.0.0.jar|HBaseIA.TwitBase.coprocessors.FollowsObserver|1001|'Updatingall regions with the new schema...1/1regions updated.Done.0 row(s) in 1.0770 secondshbase(main):003:0>enable 'follows'0 row(s) in 2.0760 seconds

当中1001为优先级。当载入多个Observer时。依照优先级次序执行。

实现Endpoint

关注人数量统计能够通过clientScan实现,相比Endpoint方案。有两个待改进点:

  1. 传输全部关注人到client,不必要的网络I/O。

  2. 拿到全部关注人Result结果后。遍历实现计数是单线程的。

实现Endpoint包含三部分

定义PRC接口

publicinterface RelationCountProtocol extends CoprocessorProtocol {    public long followedByCount(String userId) throwsIOException;}

服务端实现

和client不同,InternalScanner执行在特定Region上。返回的是原始的KeyValue对象。

packageHBaseIA.TwitBase.coprocessors;//…publicclass RelationCountImpl extends BaseEndpointCoprocessor implementsRelationCountProtocol {    @Override    public longfollowedByCount(String userId) throws IOException {        byte[]startkey = Md5Utils.md5sum(userId);        Scan scan = newScan(startkey);        scan.setFilter(newPrefixFilter(startkey));        scan.addColumn(RELATION_FAM,FROM);        scan.setMaxVersions(1);        RegionCoprocessorEnvironmentenv= (RegionCoprocessorEnvironment)getEnvironment();        InternalScanner scanner =env.getRegion().getScanner(scan);//1,server端        long sum = 0;        List<KeyValue> results= new ArrayList<KeyValue>();        boolean hasMore = false;        do {            hasMore =scanner.next(results);            sum += results.size();            results.clear();        } while (hasMore);        scanner.close();        return sum;    }}

client代码

參考凝视:

  1. 定义Call实例 
  2. 调用服务端Endpoint。
  3. 聚合全部RegionServer得到的结果

public long followedByCount (final String userId) throws Throwable {    HTableInterface followed =pool.getTable(FOLLOWED_TABLE_NAME);    final byte[] startKey = Md5Utils.md5sum(userId);    final byte[] endKey =Arrays.copyOf(startKey, startKey.length);    endKey[endKey.length-1]++;    Batch.Call<RelationCountProtocol,Long> callable =        newBatch.Call<RelationCountProtocol, Long>() {        @Override        public Longcall(RelationCountProtocol instance) throws IOException {            returninstance.followedByCount(userId);        }    };//1 call instance    Map<byte[], Long>results = followed.coprocessorExec(                                   RelationCountProtocol.class,                                   startKey,                                   endKey,                                   callable);//2 invoke endpoint    long sum = 0;    for(Map.Entry<byte[],Long> e : results.entrySet()) {        sum +=e.getValue().longValue();    }//3 aggreagte results    return sum;}

Endpoint仅仅能通过配置文件部署,还须要将相关jar包增加到HBase classpath。

<property>    <name>hbase.coprocessor.region.classes</name>    <value>HBaseIA.TwitBase.coprocessors.RelationCountImpl</value></property>

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/115767.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

  • VMware-workstation-full-12.0.1-3160714

    VMware-workstation-full-12.0.1-3160714

  • FOC和SVPWM的C语言代码实现「建议收藏」

    FOC和SVPWM的C语言代码实现「建议收藏」SVPWM的原理讲解在这儿:https://blog.csdn.net/qlexcel/article/details/74787619#comments现在开始分析C语言的代码(代码建议复制到notepad++中查看),为方便读者试验,每个代码都是独立的子模块,复制到工程中就可以编译运行:一、配置高级定时器TIM1产生6路互补PWM,带刹车保护详细配置代码如下,把下面的程序段拷贝到m…

  • Oracle 绑定变量_oracle变量不在选择列表中

    Oracle 绑定变量_oracle变量不在选择列表中begintest.test(a=>1,b=>:603);end;执行的时候报错:Ora-01008错误:Oracle?并非所有变量都已绑定原因是??b=>:603之前多了一个冒号导致他判断错误,解决方法是检查和删除不需要的冒号。…

  • 阻止mouseover冒泡行为_onmousedown是什么意思

    阻止mouseover冒泡行为_onmousedown是什么意思一.onmouseenter、onmouseoveronmouseenter事件在鼠标指针进入到绑定事件的那个元素上时触发。该事件通常与onmouseleave(在鼠标指针离开绑定事件的那个元素上时触发)事件一同使用。onmouseenter事件类似于onmouseover事件。唯一的区别是onmouseenter事件不支持冒泡。二.实例演示onmousemov…

  • dos攻击防范 java_php DOS攻击实现代码(附如何防范)

    dos攻击防范 java_php DOS攻击实现代码(附如何防范)index.php$ip=$_SERVER[‘REMOTE_ADDR’];?>PHPDoS,CodedbyEXEYourIP:(Don’tDoSyourselfnub)IP:Time:Port:AfterinitiatingtheDoSattack,pleasewaitwhilethebrowserloads.function.ph…

  • Springboot整合SSM

    Springboot整合SSM1.1创建SSM模块1.1.1系统架构图1.1.2项目结构1.1.3需求访问:http://localhost:8080/car/get返回:{“name”:”BMW”,”color”:”red”,”price”:9.9}1.1.4准备表,数据CREATETABLE`car`(`id`int(11)NOTNULLauto_increment,`name`varchar(10)defaultNULL,`color`varchar(10)d

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号