HBase总结（11）--观察者模式

Posted 2022-12-09 莫西里

一、介绍

上一届简单介绍了协处理器的主要功能以及相应的特点，但没有对写出器的具体内容作出详细的解释。这篇文章解释了协处理器的观察者模式。观察者模式就是提供了一个触发器，用户通过集成相应的类（BaseRegionObverser等），重写其中想要实现的方法，然后将协处理器加载到表中，这是表就会通过协处理器“监听”用户预先设置的动作，一旦该动作被执行，用户所写的钩子函数就被触发，然后就能实现相应的功能。

二、详解

1、观察者模式介绍

观察者模式提供了一种类似于触发器的功能，即在用户执行某些已经被指定的动作时，会触发观察者模式，并执行用户已经提前设置在观察者模式中的代码。

观察则模式主要分为三种类型：region级别的观察者模式、WAL级别的观察者模式以及master级别的观察者模式。这三种观察者模式所“监听”的对象分别是region操作级别对象、wal操作级别的对象以及master操作级别的对象。

region级别：主要提供一些针对于region级别操作（put、get、delete等）的钩子函数，用户定义的此类观察这在这些操作执行时被触发。

wal级别：主要是提供一些针对于wal级别操作的钩子函数。

master级别：主要是提供了一些针对于master级别操作（createtable、disabletable、droptable等）的钩子函数

2、region级别观察者

实现region级别的观察者时需要继承一个基本类，该类中已经包含了所有的region级别的钩子函数，用户只需要进行重写即可：BaseRegiobObverser

在提供的所有的钩子函数，都是以preDo() / postDo() 成对存在，例如 prePut() / postPut() 函数就是成对存在。preDo() 系列的函数表明了在执行Do所执行的动作之前执行钩子函数。postDo() 系列函数标明在执行了Do之后执行钩子函数。

例如下面的例子，我们要实现一个复制例子的数据，即将插入test表中的数据也全都插入到testCopy表中：

（1）进行Obverser兑现的编写

观察者实例：

import java.io.IOException;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.HBaseAdmin;
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;

/**
 * @author wozipa
 * @Date 2016-3-20 20:57
 * @see 创建两张一模一样的表
 */
public class RegionObverserExample extends BaseRegionObserver
	
	private static String tableName="testCopy";
	private static String family="copy";
	
	/**
	 * @author wozipa
	 * @Date 2016-3-20 21：00
	 * @see 进行表格的配置
	 * @return
	 */
	public Configuration init()
	
		Configuration conf=new Configuration();
		conf.set("hbase.master", "hadoop1:16000");
		conf.set("hbase.zookeeper.quorum", "hadoop1,hadoop2,hadoop3,hadoop4,hadoop5,hadoop6");
		conf.set("hbase.zookeeper.property.clientPort", "2181");
		return conf;
	
	

	/**
	 * @author wozipa
	 * @Date 2016-3-20 21:02
	 * @see 在插入数据之前将数据插入到指定的表格中
	 */
	@Override
	public void prePut(ObserverContext<RegionCoprocessorEnvironment> e, Put put, WALEdit edit, Durability durability)
			throws IOException 
		// TODO Auto-generated method stub
		Configuration conf=init();
		try 
			HBaseAdmin admin=new HBaseAdmin(conf);
			if(!admin.tableExists(tableName))
			
				HTableDescriptor tableDescriptor=new HTableDescriptor(tableName);
				HColumnDescriptor columnDescriptor=new HColumnDescriptor(family);
				tableDescriptor.addFamily(columnDescriptor);
				admin.createTable(tableDescriptor);
				admin.close();
			
			//创建数据插入该保重，保持行键值和列名、value值的相同
			Map<byte[],List<KeyValue>> map=put.getFamilyMap();
			List<KeyValue> list=map.get(put.getRow());
			Put put2=new Put(put.getRow());
			for(KeyValue kv:list)
			
				put2.addColumn(Bytes.toBytes(family),kv.getQualifier(),kv.getValue());
			
			//创建表连接
			HTable table=new HTable(conf, tableName);
			table.put(put2);
			//
			table.close();
		 catch (Exception e1) 
			// TODO: handle exception
			e1.printStackTrace();
		
	

在该对象中，通过实现prePut() 函数，即在test表的region执行put方法之前，将put中的数据插入到testCopy中。

（2）将项目打包成jar包后上传到HDFS上

然后将该类所在的项目打包上传到HDFS上：

（3）编写例子类

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Coprocessor;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

import com.yammer.metrics.stats.EWMA;

/**
 * @author wozipa
 * @Date 2016-3-20 21:16
 * @see 用来测试协处理器的各个功能的函数集合
 *
 */
public class Main 
	
	private static String tableName="test";
	private static String family="test";
	
	public static void main(String[] args)
	
		Main main=new Main();
		main.putCoprocessExample();
	
	
	public Configuration init()
	
		Configuration conf=new Configuration();
		conf.set("hbase.master", "hadoop1:16000");
		conf.set("hbase.zookeeper.quorum", "hadoop1,hadoop2,hadoop3,hadoop4,hadoop5,hadoop6");
		conf.set("hbase.zookeeper.property.clientPort", "2181");
		return conf;
	
	
	/**
	 * @author wozipa
	 * @Date 2016-3-20 21：17
	 * @see 进行数据
	 */
	public void putCoprocessExample()
	
		Configuration conf=init();
		HBaseAdmin admin=null;
		try 
			admin=new HBaseAdmin(conf);
			if(!admin.tableExists(tableName))
			
				HTableDescriptor tableDescriptor=new HTableDescriptor(tableName);
				HColumnDescriptor columnDescriptor=new HColumnDescriptor(family);
				tableDescriptor.addFamily(columnDescriptor);
				Path path=new Path("hdfs://192.98.12.234:9000/Coprocessor/Copro.jar");
				tableDescriptor.setValue("COPROCESSOR$1",path.toString()+"|"+"com.geoway.coprocessor.RegionObverserExample|"+Coprocessor.PRIORITY_USER);
				admin.createTable(tableDescriptor);
			
			//想数据库中插入数据
			Put put=new Put(Bytes.toBytes("row-2"));
			put.addColumn(Bytes.toBytes(family), Bytes.toBytes("qualifier1"), Bytes.toBytes("value2"));
			HTable table=new HTable(conf, tableName);
			table.put(put);
			table.close();
		 catch (IOException e) 
			// TODO Auto-generated catch block
			e.printStackTrace();
		
	

需要注意的是，在定义test表时，需要在表结构中定义协处理器的信息。以为是通过代码添加表协处理器，因此该协处理器只针对test表有效。

最后执行结构会发现两个表同时被创建：

3、master级别的观察者模式

master级别的观察者模式与region级别的模式相同，只不过其“监听”的对象变成了master级别的操作。也就是相当于SQL语句中的DDL语句。主要包括对表的一些操作对象和region级别的操作钩子函数。

HBase中也提供了一个 BaseMasterObverser对象，该对象中也封装了所有的DDL操作的钩子函数。用户只需要继承该对象，并重写相应的方法，就可实现相应的功能。其实现流程与region级别的流程一样。

三、总结

观察者模式提供了用户在进行region级别操作之前与之后的自定义动作，因此可以实现很多功能，例如可以听过观察者实现二级索引等功能。