HBase Coprocessor 之 RegionObserver(hbase 0.96.0)

关于hbase的协处理器（Coprocessor ）的介绍在此不做赘述，本文基于RegionObserver实现hbase的计数器，用于实时统计pv等关键指标。

一、背景介绍：

  RegionObserver：hbase协处理器的一种，类似于关系型数据库的存储过程；

 Counter：hbase的计数器，线程安全，本质上就是一张表

二、实现原理

   使用hbase的触发器，当执行put操作后，计数器加一；执行delete操作后，计数器减一；

三、代码

package com.cxk.coprocessor.test;

import java.io.IOException;

import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
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.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

public class TestRegionObserver extends BaseRegionObserver{

@Override
public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e,
Delete delete, WALEdit edit, Durability durability)
throws IOException {
// TODO Auto-generated method stub
super.postDelete(e, delete, edit, durability);
HTable table=new HTable(e.getEnvironment().getConfiguration(),"sumcounter");
table.incrementColumnValue(Bytes.toBytes("row1"), Bytes.toBytes("cf"), Bytes.toBytes("count"),(long)-1);
table.close();

}

@Override
public void postPut(ObserverContext<RegionCoprocessorEnvironment> e,
Put put, WALEdit edit, Durability durability) throws IOException {
// TODO Auto-generated method stub
super.postPut(e, put, edit, durability);
HTable table=new HTable(e.getEnvironment().getConfiguration(),"sumcounter");
table.incrementColumnValue(Bytes.toBytes("row1"), Bytes.toBytes("cf"), Bytes.toBytes("count"),(long)1);
table.close();

}

}

如果一次更新多条记录，参考以下代码：

Increment increment=new Increment(Bytes.toBytes(row));
increment.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), (long)2);
increment.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier2), (long)3);
Result result=table.increment(increment);

四、部署

详见：HBase Coprocessor 之 endpiont(hbase 0.96.0)