`
zhang_xzhi_xjtu
  • 浏览: 535009 次
  • 性别: Icon_minigender_1
  • 来自: 杭州
社区版块
存档分类
最新评论

hbase的CoprocessorProtocol及一个简单的通用扩展实现V2

 
阅读更多
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 CommonCoprocessorProtocol extends CoprocessorProtocol {

	public static final long VERSION = 345L;

	public <T> T handle(KeyValueListHandler<T> handler, Scan scan)
			throws IOException;

}


定义该protocol的实现。
public class CommonEndpointImpl extends BaseEndpointCoprocessor implements
		CommonCoprocessorProtocol {

	protected static Log log = LogFactory.getLog(CommonEndpointImpl.class);

	@Override
	public ProtocolSignature getProtocolSignature(String protocol,
			long version, int clientMethodsHashCode) throws IOException {
		if (CommonCoprocessorProtocol.class.getName().equals(protocol)) {
			return new ProtocolSignature(CommonCoprocessorProtocol.VERSION,
					null);
		}
		throw new IOException("Unknown protocol: " + protocol);
	}

	@Override
	public <T> T handle(KeyValueListHandler<T> handler, Scan scan)
			throws IOException {

		InternalScanner scanner = ((RegionCoprocessorEnvironment) getEnvironment())
				.getRegion().getScanner(scan);
		List<KeyValue> results = new ArrayList<KeyValue>();
		T t = handler.getInitValue();
		try {
			boolean hasMoreRows = false;
			do {
				hasMoreRows = scanner.next(results);
				t = handler.handle(results, t);
				results.clear();
			} while (hasMoreRows);
		} finally {
			scanner.close();
		}
		return t;
	}
}


定义一个KeyValueListHandler。
public interface KeyValueListHandler<T> extends Writable {

	public T getInitValue();

	public T handle(List<KeyValue> keyValues, T t);
}


定义一个reduce。
public interface ClientReducer<T, R> {

	public R getInitValue();

	public R reduce(R r, T t);
}


定义一个client。
public class CpClient {

	private HTableInterface table;

	public CpClient(HTableInterface table) {
		this.table = table;
	}

	public <T, R> R call(final KeyValueListHandler<T> handler,
			final ClientReducer<T, R> reducer, final Scan scan)
			throws Throwable {

		class MyCallBack implements Batch.Callback<T> {
			R r = reducer.getInitValue();

			R getResult() {
				return r;
			}

			@Override
			public synchronized void update(byte[] region, byte[] row, T result) {
				r = reducer.reduce(r, result);
			}
		}

		MyCallBack myCallBack = new MyCallBack();

		try {
			table.coprocessorExec(CommonCoprocessorProtocol.class,
					scan.getStartRow(), scan.getStopRow(),
					new Batch.Call<CommonCoprocessorProtocol, T>() {
						@Override
						public T call(CommonCoprocessorProtocol instance)
								throws IOException {
							return instance.handle(handler, scan);
						}
					}, myCallBack);
		} finally {
			table.close();
		}

		return myCallBack.getResult();
	}
}


这样,我们就有了一个protocol的通用框架。

假设我们要同时得到多个列的sum和结果的count,我们通过实现这些接口和定义一些request和result类来实现。
public class AggrRequest implements Writable {

	private List<byte[]> families = new ArrayList<byte[]>();
	private List<byte[]> qualifiers = new ArrayList<byte[]>();

	public AggrRequest() {
	}

	public void add(String family, String qualifier) {
		if (family != null && qualifier != null) {
			this.families.add(Bytes.toBytes(family));
			this.qualifiers.add(Bytes.toBytes(qualifier));
		}
	}

	public int getColumnSize() {
		return families.size();
	}

	public byte[] getFamily(int index) {
		return families.get(index);
	}

	public byte[] getQualifer(int index) {
		return qualifiers.get(index);
	}

	@Override
	public void readFields(DataInput dataInput) throws IOException {

		int size = dataInput.readInt();

		for (int i = 0; i < size; i++) {
			families.add(Bytes.toBytes(dataInput.readUTF()));
		}

		for (int i = 0; i < size; i++) {
			qualifiers.add(Bytes.toBytes(dataInput.readUTF()));
		}

	}

	@Override
	public void write(DataOutput dataOutput) throws IOException {

		dataOutput.writeInt(getColumnSize());

		for (byte[] b : families) {
			dataOutput.writeUTF(Bytes.toString(b));
		}

		for (byte[] b : qualifiers) {
			dataOutput.writeUTF(Bytes.toString(b));
		}
	}
}

public class AggrResult implements Writable {
	private AggrRequest aggrRequest;
	private long[] sum;
	private long count;

	public AggrResult() {
	}

	public AggrResult(AggrRequest aggrRequest) {
		this.aggrRequest = aggrRequest;
		sum = new long[aggrRequest.getColumnSize()];
	}

	public int getColumnSize() {
		return aggrRequest.getColumnSize();
	}

	public byte[] getFamily(int index) {
		return aggrRequest.getFamily(index);
	}

	public byte[] getQualifer(int index) {
		return aggrRequest.getQualifer(index);
	}

	public long getSum(int index) {
		return sum[index];
	}

	public void setSum(int index, long value) {
		sum[index] = value;
	}

	// getter and setter.
	public long getCount() {
		return count;
	}

	public void setCount(long count) {
		this.count = count;
	}

	@Override
	public void readFields(DataInput dataInput) throws IOException {
		int columnSize = dataInput.readInt();

		sum = new long[columnSize];

		for (int i = 0; i < columnSize; i++) {
			sum[i] = dataInput.readLong();
		}

		count = dataInput.readLong();

		aggrRequest = new AggrRequest();

		aggrRequest.readFields(dataInput);
	}

	@Override
	public void write(DataOutput dataOutput) throws IOException {

		dataOutput.writeInt(aggrRequest.getColumnSize());

		for (long v : sum) {
			dataOutput.writeLong(v);
		}

		dataOutput.writeLong(count);

		aggrRequest.write(dataOutput);
	}

}



public class AggrHandler implements KeyValueListHandler<AggrResult> {

	private AggrRequest aggrRequest;

	public AggrHandler() {
	}

	public AggrHandler(AggrRequest aggrRequest) {
		this.aggrRequest = aggrRequest;
	}

	@Override
	public void readFields(DataInput dataInput) throws IOException {
		aggrRequest = new AggrRequest();
		aggrRequest.readFields(dataInput);
	}

	@Override
	public void write(DataOutput dataOutput) throws IOException {
		aggrRequest.write(dataOutput);
	}

	@Override
	public AggrResult getInitValue() {
		AggrResult aggrResult = new AggrResult(aggrRequest);
		return aggrResult;
	}

	@Override
	public AggrResult handle(List<KeyValue> keyValues, AggrResult t) {
		if (keyValues.isEmpty()) {
			return t;
		}

		t.setCount(t.getCount() + 1);

		int columnSize = t.getColumnSize();
		for (int i = 0; i < columnSize; i++) {
			byte[] family = t.getFamily(i);
			byte[] qualifer = t.getQualifer(i);
			for (KeyValue kv : keyValues) {
				if (kv != null) {
					if (Bytes.equals(qualifer, 0, qualifer.length,
							kv.getBuffer(), kv.getQualifierOffset(),
							kv.getQualifierLength())
							&& Bytes.equals(family, 0, family.length,
									kv.getBuffer(), kv.getFamilyOffset(),
									kv.getFamilyLength())) {

						if (kv.getValueLength() == Bytes.SIZEOF_LONG) {
							long tem = Bytes.toLong(kv.getBuffer(),
									kv.getValueOffset());
							t.setSum(i, t.getSum(i) + tem);
						}

					}
				}
			}
		}

		return t;
	}
}




public class AggrReducer implements ClientReducer<AggrResult, AggrResult> {

	@Override
	public AggrResult getInitValue() {
		return null;
	}

	@Override
	public AggrResult reduce(AggrResult r, AggrResult t) {
		if (r == null)
			return t;
		if (t == null)
			return r;
		r.setCount(r.getCount() + t.getCount());
		int columnSize = r.getColumnSize();
		for (int i = 0; i < columnSize; i++) {
			r.setSum(i, r.getSum(i) + t.getSum(i));
		}
		return r;
	}

}


有了CoprocessorProtocol,可以扩展出来很多的功能,这个机制还是很强大的。

代码见https://github.com/zhang-xzhi/simplehbase
并且有测试代码。
分享到:
评论

相关推荐

    hbase分页查询实现.pdf

    HBase作为一个NoSQL数据库,具有高性能、高可扩展性和高可靠性等特点,但是在查询方面却存在一些限制,例如不支持分页查询。这就使得开发者需要自己实现分页查询功能。本文将讲解如何使用Java语言实现HBase的分页...

    HbaseTemplate 操作hbase

    在IT行业中,尤其是在大数据处理领域,HBase是一个广泛使用的分布式、高性能、列式存储的NoSQL数据库。HBase是建立在Hadoop文件系统(HDFS)之上,为处理大规模数据提供了一个高效的数据存储解决方案。而Spring Data...

    hbase分页查询实现[归类].pdf

    HBase分页查询实现 HBase是一种基于分布式的NoSQL数据库,它提供了高效的数据存储和检索能力。然而,HBase本身不支持分页查询,...本文提供了一个简单的示例代码,旨在帮助开发者更好地理解HBase分页查询的实现机制。

    hbase用于查询客户端工具

    6. **HBase Console**:这是HBase提供的一个简单的Web界面,用于查看表的信息、region分布和集群状态。它不支持复杂的查询,但对于快速检查和监控HBase实例非常有用。 7. **HBase MapReduce**:MapReduce是Hadoop...

    Hbase是Apache的NoSQL分布式可扩展Hadoop数据库,可以很好地横向扩展.rar

    Hbase是Apache的NoSQL分布式可扩展Hadoop数据库,可以很好地横向扩展。Hbase中的数据是面向列的数据库,其中结构化数据存储在键值对中。Hbase用Java编写。Hbase的灵感来自Google Paper-“大表:结构化数据的分布式...

    HBase_SI_--_实现HBase_ACID的理论

    根据给定文件的信息,本文将深入探讨"HBase_SI"这一理论框架,该理论旨在实现HBase中的ACID特性。文章将从多个角度分析...对于那些需要高度一致性和可靠性的大数据应用来说,HBase_SI无疑是一个非常有价值的解决方案。

    HBase实现批量存取

    本项目实现了在Eclipse环境下对HBase的批量存取操作,这对于理解HBase的工作原理以及如何在实际应用中使用HBase具有重要的参考价值。 首先,我们需要了解HBase的基本概念。HBase是建立在Hadoop文件系统(HDFS)之上...

    Hbase的安装过程及基本操作

    接下来,编写一个简单的Java程序,如`ExampleForHBase`,它展示了如何连接到Hbase,创建表,插入数据,以及获取数据。代码示例中,我们创建了一个名为"student"的表,包含一个列族"score",并插入了张三的各科成绩...

    Hbase权威指南(HBase: The Definitive Guide)

    - **HBase的历史**:HBase起源于一个叫做Hadoop的项目中的子项目,最初是为了实现一个类似于Bigtable的功能而创建的。 - **术语介绍**: - **背景层**(Backdrop):HBase运行于Hadoop之上,利用Hadoop提供的分布式...

    hbase-1.1.5-bin版本的压缩包,下载到本地解压后即可使用 HBase 是一个开源的、分布式的NoSQL数据库

    HBase 是一个开源的、分布式的、版本化的 NoSQL 数据库(也即非关系型数据库),它利用 Hadoop 分布式文件系统(Hadoop Distributed File System,HDFS)提供分布式数据存储。与传统的关系型数据库类似,HBase 也以...

    基于springboot集成hbase过程解析

    HBaseTemplate是SpringBoot提供的一个模板类,用于简化HBase的操作。下面是一个使用HBaseTemplate的示例: ```java @Service @Slf4j public class HBaseService { @Autowired private HbaseTemplate hbase...

    HBase分布式事务与SQL实现

    从技术角度看,HBase分布式事务与SQL实现是一个复杂的话题,它涉及到分布式系统理论、事务处理机制、数据库设计和实际的应用场景等多个领域。理解和掌握这些知识点对于想要深入研究和使用HBase以及TiDB这类分布式...

    HBase学习利器:HBase实战

    - **第2章:入门指南**:通过一个简单的例子来展示如何安装配置HBase环境,以及如何使用命令行工具进行基本操作,如创建表、插入数据和查询数据。 - **第3章:分布式HBase、HDFS和MapReduce**:深入探讨HBase如何...

    Hbase java DBHelper CRUD等通用方法

    节制2013年12月23日最新的hadoop和hbase兼容版本搭建 hadoop-2.2.0 hbase-0.96.1.1 java Hbase java DBHelper CRUD等通用方法 花了两天时间整理的,含有lib包 & 源码

    hbase社区2018精选资料

    HBase是一个高性能的开源NoSQL数据库,属于BigTable的开源实现,其分布式、多版本、面向列的特点使其适合存储和处理大量的非结构化数据。随着大数据技术的发展,HBase面临着更复杂多样的数据格式和业务需求,因此...

    hbase存储csv数据的代码实现

    以下是一个简单的Python示例,展示如何使用HappyBase库(一个Python HBase客户端)来实现这个过程: ```python import happybase # 连接到HBase connection = happybase.Connection('localhost') # 创建表 table ...

    hbase数据可视化系统

    SpringBoot是Spring框架的一个扩展,旨在简化Spring应用的初始搭建以及开发过程。它集成了大量常用的第三方库配置,如JDBC、MongoDB、JPA、RabbitMQ、Quartz等,只需少量配置就能创建一个独立的、生产级别的基于...

Global site tag (gtag.js) - Google Analytics