简单之美 | HBase Thrift客户端Java API实践

HBase的Thrift API定义，可以通过链接 http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/resources/org/apache/hadoop/hbase/thrift/Hbase.thrift?view=markup看到，我们需要安装Thrift编译器，才能生成HBase跨语言的API。
首先下载上面链接的内容，保存为Hbase.thrift。
然后，执行如下命令，生成不同编程语言的HBase API：

[hadoop@master hbase]$ thrift --gen cpp Hbase.thrift
[hadoop@master hbase]$ thrift --gen java Hbase.thrift
[hadoop@master hbase]$ thrift --gen py Hbase.thrift
[hadoop@master hbase]$ thrift --gen perl Hbase.thrift
[hadoop@master hbase]$ thrift --gen csharp Hbase.thrift
[hadoop@master hbase]$ thrift --gen php Hbase.thrift
[hadoop@master hbase]$ thrift --gen js Hbase.thrift
[hadoop@master hbase]$ thrift --gen go Hbase.thrift
[hadoop@master hbase]$ thrift --gen erl Hbase.thrift
[hadoop@master hbase]$ thrift --gen delphi Hbase.thrift
[hadoop@master hbase]$ thrift --gen hs Hbase.thrift
[hadoop@master hbase]$ thrift --gen html Hbase.thrift
[hadoop@master hbase]$ thrift --gen c_glib Hbase.thrift
[hadoop@master hbase]$ thrift --gen cocoa Hbase.thrift
[hadoop@master hbase]$ thrift --gen rb Hbase.thrift
[hadoop@master hbase]$ thrift --gen st Hbase.thrift
[hadoop@master hbase]$ thrift --gen xsd Hbase.thrift
[hadoop@master hbase]$ ls
gen-as3     gen-cocoa  gen-csharp  gen-erl  gen-hs    gen-java  gen-perl  gen-py  gen-st   Hbase.thrift
gen-c_glib  gen-cpp    gen-delphi  gen-go   gen-html  gen-js    gen-php   gen-rb  gen-xsd

这里，我们基于Java语言，使用HBase 的Thrift 客户端API访问HBase表。事实上，如果使用Java来实现对HBase表的操作，最好是使用HBase的原生API，无论从性能还是便利性方面，都会提供更好的体验。使用Thrift API访问，实际也是在HBase API之上进行了一层封装，可能初次使用Thrift API感觉很别扭，有时候还要参考Thrift服务端的实现代码。
准备工作如下：

下载Thrift软件包，解压缩后，拷贝thrift-0.9.0/lib/java/src下面的代码到工作区（开发工具中）
将上面生成的gen-java目录中代码拷贝到工作区
保证HBase集群正常运行，接着启动HBase的Thrift服务，执行如下命令：

bin/hbase thrift -b master -p 9090 start

上面，HBase的Thrift服务端口为9090，下面通过Thrift API访问的时候，需要用到，而不是HBase的服务端口（默认60000）。
接着，实现一个简单的例子，访问Hbase表。
首先，我们通过HBase Shell创建一个表：

create 'test_info', 'info'

表名为test_info，列簇名称为info。
然后，我们开始基于上面生成的Thrift代码来实现对HBase表的操作。
我们在客户端，进行了一层抽象，更加便于传递各种参数，抽象类为AbstractHBaseThriftService，代码如下所示：

package org.shirdrn.cloud.hbase.thrift;

import java.util.List;
import java.util.Map;

import org.apache.hadoop.hbase.thrift.generated.Hbase;
import org.apache.hadoop.hbase.thrift.generated.TRowResult;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;

public abstract class AbstractHBaseThriftService {

	protected static final String CHARSET = "UTF-8";
	private String host = "localhost";
	private int port = 9090;
	private final TTransport transport;
	protected final Hbase.Client client;

	public AbstractHBaseThriftService() {
		transport = new TSocket(host, port);
		TProtocol protocol = new TBinaryProtocol(transport, true, true);
		client = new Hbase.Client(protocol);
	}

	public AbstractHBaseThriftService(String host, int port) {
		super();
		transport = new TSocket(host, port);
		TProtocol protocol = new TBinaryProtocol(transport, true, true);
		client = new Hbase.Client(protocol);
	}

	public void open() throws TTransportException {
		if(transport != null) {
			transport.open();
		}
	}

	public void close() {
		if(transport != null) {
			transport.close();
		}
	}
	
	public abstract List<String> getTables() throws TException;
	
	public abstract void update(String table, String rowKey, boolean writeToWal,
			String fieldName, String fieldValue, Map<String, String> attributes) throws TException;
	public abstract void update(String table, String rowKey, boolean writeToWal,
			Map<String, String> fieldNameValues, Map<String, String> attributes) throws TException;
	
	public abstract void deleteCell(String table, String rowKey, boolean writeToWal,
			String column, Map<String, String> attributes) throws TException;
	public abstract void deleteCells(String table, String rowKey, boolean writeToWal,
			List<String> columns, Map<String, String> attributes) throws TException;
	
	 public abstract void deleteRow(String table, String rowKey,
		        Map<String, String> attributes) throws TException;
		        
	public abstract int scannerOpen(String table, String startRow, List<String> columns,
	        Map<String, String> attributes) throws TException;
	public abstract int scannerOpen(String table, String startRow, String stopRow, List<String> columns,
	        Map<String, String> attributes) throws TException;
	public abstract int scannerOpenWithPrefix(String table, String startAndPrefix,
            List<String> columns, Map<String, String> attributes) throws TException;
	public abstract int scannerOpenTs(String table, String startRow,
	        List<String> columns, long timestamp, Map<String, String> attributes) throws TException;
	public abstract int scannerOpenTs(String table, String startRow, String stopRow,
	        List<String> columns, long timestamp, Map<String, String> attributes) throws TException;
		        
	public abstract List<TRowResult> scannerGetList(int id, int nbRows) throws TException;
	public abstract List<TRowResult> scannerGet(int id) throws TException;
	
	public abstract List<TRowResult> getRow(String table, String row,
		        Map<String, String> attributes) throws TException;
	public abstract List<TRowResult> getRows(String table,
             List<String> rows, Map<String, String> attributes) throws TException;
	public abstract List<TRowResult> getRowsWithColumns(String table,
             List<String> rows, List<String> columns, Map<String, String> attributes) throws TException;
	
	public abstract void scannerClose(int id) throws TException;
	
	/**
	 * Iterate result rows(just for test purpose)
	 * @param result
	 */
	public abstract void iterateResults(TRowResult result);

}

这里，简单叙述一下，我们提供的客户端API的基本功能：

建立到Thrift服务的连接：open()
获取到HBase中的所有表名：getTables()
更新HBase表记录：update()
删除HBase表中一行的记录的数据（cell）：deleteCell()和deleCells()
删除HBase表中一行记录：deleteRow()
打开一个Scanner，返回id：scannerOpen()、scannerOpenWithPrefix()和scannerOpenTs()；然后用返回的id迭代记录：scannerGetList()和scannerGet()
获取一行记录结果：getRow()、getRows()和getRowsWithColumns()
关闭一个Scanner：scannerClose()
迭代结果，用于调试：iterateResults()

比如，我们想要实现分页的逻辑，可能和传统的关系型数据库操作有些不同。基于HBase表的实现是，首先打开一个Scanner实例（例如调用scannerOpen()），返回一个id，然后再使用该id，调用scannerGetList()方法（可以指定每次返回几条记录的变量nbRows的值），返回一个记录列表，反复调用该scannerGetList()方法，直到此次没有结果返回为止。后面会通过测试用例来实际体会。
现在，我们基于上抽象出来的客户端操作接口，给出一个基本的实现，代码如下所示：

package org.shirdrn.cloud.hbase.thrift;

import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.hbase.thrift.generated.IOError;
import org.apache.hadoop.hbase.thrift.generated.Mutation;
import org.apache.hadoop.hbase.thrift.generated.TCell;
import org.apache.hadoop.hbase.thrift.generated.TRowResult;
import org.apache.thrift.TException;


public class HBaseThriftClient extends AbstractHBaseThriftService {

	public HBaseThriftClient() {
		super();
	}
	
	public HBaseThriftClient(String host, int port) {
		super(host, port);
	}
	
	@Override
	public List<String> getTables() throws TException {
		List<String> list = new ArrayList<String>(0);
		for (ByteBuffer buf : client.getTableNames()) {
			byte[] name = decode(buf);
			list.add(new String(name));
		}
		return list;
	}
	
	static ByteBuffer wrap(String value) {
		ByteBuffer bb = null;
		try {
			bb = ByteBuffer.wrap(value.getBytes(CHARSET));
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		}
		return bb;
	}
	
	protected byte[] decode(ByteBuffer buffer) {
		byte[] bytes = new byte[buffer.limit()];
		for (int i = 0; i < buffer.limit(); i++) {
			bytes[i] = buffer.get();
		}
		return bytes;
	}

	@Override
	public void update(String table, String rowKey, boolean writeToWal,
			Map<String, String> fieldNameValues, Map<String, String> attributes) throws TException {
		List<Mutation> mutations = new ArrayList<Mutation>();
		for(Map.Entry<String, String> entry : fieldNameValues.entrySet()) {
			mutations.add(new Mutation(false, wrap(entry.getKey()), wrap(entry.getValue()), writeToWal));
		}
		Map<ByteBuffer, ByteBuffer> wrappedAttributes = encodeAttributes(attributes);
		ByteBuffer tableName = wrap(table);
		ByteBuffer row = wrap(rowKey);
		client.mutateRow(tableName, row, mutations, wrappedAttributes);
	}

	@Override
	public void update(String table, String rowKey, boolean writeToWal, 
			String fieldName, String fieldValue, Map<String, String> attributes) throws IOError, TException {
		Map<String, String> fieldNameValues = new HashMap<String, String>();
		fieldNameValues.put(fieldName, fieldValue);
		update(table, rowKey, writeToWal, fieldNameValues, attributes);
	}
	
	
	@Override
	public void deleteCells(String table, String rowKey, boolean writeToWal, 
			List<String> columns, Map<String, String> attributes) throws TException {
		List<Mutation> mutations = new ArrayList<Mutation>();
		for(String column : columns) {
			mutations.add(new Mutation(false, wrap(column), null, writeToWal));
		}
		Map<ByteBuffer, ByteBuffer> wrappedAttributes = encodeAttributes(attributes);
		ByteBuffer tableName = wrap(table);
		ByteBuffer row = wrap(rowKey);
		client.mutateRow(tableName, row, mutations, wrappedAttributes);
	}

	@Override
	public void deleteCell(String table, String rowKey, boolean writeToWal, 
			String column, Map<String, String> attributes) throws TException {
		List<String> columns = new ArrayList<String>(1);
		columns.add(column);
		deleteCells(table, rowKey, writeToWal, columns, attributes);
	}
	
	
	@Override
	public void deleteRow(String table, String rowKey, Map<String, String> attributes) throws TException {
		ByteBuffer tableName = wrap(table);
		ByteBuffer row = wrap(rowKey);
		Map<ByteBuffer, ByteBuffer> wrappedAttributes = encodeAttributes(attributes);
		client.deleteAllRow(tableName, row, wrappedAttributes);
	}
	
	
	@Override
	public int scannerOpen(String table, String startRow, List<String> columns, 
			Map<String, String> attributes) throws TException {
		ByteBuffer tableName = wrap(table);
		List<ByteBuffer> fl = encodeColumns(columns);
		Map<ByteBuffer, ByteBuffer> wrappedAttributes = encodeAttributes(attributes);
		return client.scannerOpen(tableName, wrap(startRow), fl, wrappedAttributes);
	}

	@Override
	public int scannerOpen(String table, String startRow, String stopRow, List<String> columns, 
			Map<String, String> attributes) throws TException {
		ByteBuffer tableName = wrap(table);
		List<ByteBuffer> fl = encodeColumns(columns);
		Map<ByteBuffer, ByteBuffer> wrappedAttributes = encodeAttributes(attributes);
		return client.scannerOpenWithStop(tableName, wrap(startRow), wrap(stopRow), fl, wrappedAttributes);
	}

	@Override
	public int scannerOpenWithPrefix(String table, String startAndPrefix, List<String> columns, 
			Map<String, String> attributes) throws TException {
		ByteBuffer tableName = wrap(table);
		List<ByteBuffer> fl = encodeColumns(columns);
		Map<ByteBuffer, ByteBuffer> wrappedAttributes = encodeAttributes(attributes);
		return client.scannerOpenWithPrefix(tableName, wrap(startAndPrefix), fl, wrappedAttributes);
	}

	@Override
	public int scannerOpenTs(String table, String startRow, List<String> columns, 
			long timestamp, Map<String, String> attributes) throws TException {
		ByteBuffer tableName = wrap(table);
		List<ByteBuffer> fl = encodeColumns(columns);
		Map<ByteBuffer, ByteBuffer> wrappedAttributes = encodeAttributes(attributes);
		return client.scannerOpenTs(tableName, wrap(startRow), fl, timestamp, wrappedAttributes);
	}

	@Override
	public int scannerOpenTs(String table, String startRow, String stopRow, List<String> columns, 
			long timestamp, Map<String, String> attributes) throws TException {
		ByteBuffer tableName = wrap(table);
		List<ByteBuffer> fl = encodeColumns(columns);
		Map<ByteBuffer, ByteBuffer> wrappedAttributes = encodeAttributes(attributes);
		return client.scannerOpenWithStopTs(tableName, wrap(startRow), wrap(stopRow), fl, timestamp, wrappedAttributes);
	}

	@Override
	public List<TRowResult> scannerGetList(int id, int nbRows) throws TException {
		return client.scannerGetList(id, nbRows);
	}

	@Override
	public List<TRowResult> scannerGet(int id) throws TException {
		return client.scannerGetList(id, 1);
	}
	
	@Override
	public List<TRowResult> getRow(String table, String row, 
			Map<String, String> attributes) throws TException {
		ByteBuffer tableName = wrap(table);
		Map<ByteBuffer, ByteBuffer> wrappedAttributes = encodeAttributes(attributes);
		return client.getRow(tableName, wrap(row), wrappedAttributes);
	}

	@Override
	public List<TRowResult> getRows(String table, List<String> rows, 
			Map<String, String> attributes) throws TException {
		ByteBuffer tableName = wrap(table);
		Map<ByteBuffer, ByteBuffer> wrappedAttributes = encodeAttributes(attributes);
		List<ByteBuffer> wrappedRows = encodeRows(rows);
		return client.getRows(tableName, wrappedRows, wrappedAttributes);
	}

	@Override
	public List<TRowResult> getRowsWithColumns(String table, List<String> rows, 
			List<String> columns, Map<String, String> attributes) throws TException {
		ByteBuffer tableName = wrap(table);
		List<ByteBuffer> wrappedRows = encodeRows(rows);
		List<ByteBuffer> wrappedColumns = encodeColumns(columns);
		Map<ByteBuffer, ByteBuffer> wrappedAttributes = encodeAttributes(attributes);
		return client.getRowsWithColumns(tableName, wrappedRows, wrappedColumns, wrappedAttributes);
	}
	
	private List<ByteBuffer> encodeColumns(List<String> columns) {
		List<ByteBuffer> fl = new ArrayList<ByteBuffer>(0);
		for(String column : columns) {
			fl.add(wrap(column));
		}
		return fl;
	}
	
	private Map<ByteBuffer, ByteBuffer> encodeAttributes(Map<String, String> attributes) {
		Map<ByteBuffer, ByteBuffer> wrappedAttributes = null;
		if(attributes != null && !attributes.isEmpty()) {
			wrappedAttributes = new HashMap<ByteBuffer, ByteBuffer>(1);
			for(Map.Entry<String, String> entry : attributes.entrySet()) {
				wrappedAttributes.put(wrap(entry.getKey()), wrap(entry.getValue()));
			}
		}
		return wrappedAttributes;
	}
	
	private List<ByteBuffer> encodeRows(List<String> rows) {
		List<ByteBuffer> list = new ArrayList<ByteBuffer>(0);
		for(String row : rows) {
			list.add(wrap(row));
		}
		return list;
	}
	
	@Override
	public void iterateResults(TRowResult result) {
		Iterator<Entry<ByteBuffer, TCell>> iter = result.columns.entrySet().iterator();
		System.out.println("RowKey=" + new String(result.getRow()));
		while (iter.hasNext()) {
			Entry<ByteBuffer, TCell> entry = iter.next();
			System.out.println("\tCol=" + new String(decode(entry.getKey())) + ", Value=" + new String(entry.getValue().getValue()));
		}
	}
	
	@Override
	public void scannerClose(int id) throws TException {
		client.scannerClose(id);		
	}
}

上面代码，给出了基本的实现，接着我们给出测试用例，调用我们实现的客户端操作，与HBase表进行交互。实现的测试用例类如下所示：

package org.shirdrn.cloud.hbase.thrift;

import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;

import org.apache.hadoop.hbase.thrift.generated.IOError;
import org.apache.hadoop.hbase.thrift.generated.IllegalArgument;
import org.apache.hadoop.hbase.thrift.generated.TRowResult;
import org.apache.thrift.TException;
import org.apache.thrift.transport.TTransportException;

public class Test {

	private static final String CHARSET = "UTF-8";
	static DecimalFormat formatter = new DecimalFormat("00");
	private final AbstractHBaseThriftService client;
	
	public Test(String host, int port) {
		client = new HBaseThriftClient(host, port);
		try {
			client.open();
		} catch (TTransportException e) {
			e.printStackTrace();
		}
	}
	
	public Test() {
		this("master", 9090);
	}
	
	static String randomlyBirthday() {
		Random r = new Random();
		int year = 1900 + r.nextInt(100);
		int month = 1 + r.nextInt(12);
		int date = 1 + r.nextInt(30);
		return String.valueOf(year + "-" + formatter.format(month) + "-" + formatter.format(date));
	}
	
	static String randomlyGender() {
		Random r = new Random();
		int flag = r.nextInt(2);
		return flag == 0 ? "M" : "F";
	}
	
	static String randomlyUserType() {
		Random r = new Random();
		int flag = 1 + r.nextInt(10);
		return String.valueOf(flag);
	}
	
	static ByteBuffer wrap(String value) {
		ByteBuffer bb = null;
		try {
			bb = ByteBuffer.wrap(value.getBytes(CHARSET));
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		}
		return bb;
	}
	
	static DecimalFormat rowKeyFormatter = new DecimalFormat("0000");
	
	public void caseForUpdate() throws TException {
		boolean writeToWal = false;
		Map<String, String> attributes = new HashMap<String, String>(0);
		String table = setTable();
		// put kv pairs
		for (long i = 0; i < 10000; i++) {
			String rowKey = rowKeyFormatter.format(i);
			Map<String, String> fieldNameValues = new HashMap<String, String>();
			fieldNameValues.put("info:birthday", randomlyBirthday());
			fieldNameValues.put("info:user_type", randomlyUserType());
			fieldNameValues.put("info:gender", randomlyGender());
			client.update(table, rowKey, writeToWal, fieldNameValues, attributes);
		}
	}
	
	public void caseForDeleteCells() throws TException {
		boolean writeToWal = false;
		Map<String, String> attributes = new HashMap<String, String>(0);
		String table = setTable();
		// put kv pairs
		for (long i = 5; i < 10; i++) {
			String rowKey = rowKeyFormatter.format(i);
			List<String> columns = new ArrayList<String>(0);
			columns.add("info:birthday");
			client.deleteCells(table, rowKey, writeToWal, columns, attributes);
		}
	}

	private String setTable() {
		String table = "test_info";
		return table;
	}
	
	public void caseForDeleteRow() throws TException {
		Map<String, String> attributes = new HashMap<String, String>(0);
		String table = setTable();
		// delete rows
		for (long i = 5; i < 10; i++) {
			String rowKey = rowKeyFormatter.format(i);
			client.deleteRow(table, rowKey, attributes);
		}
	}
	
	public void caseForScan() throws TException {
		Map<String, String> attributes = new HashMap<String, String>(0);
		String table = setTable();
		String startRow = "0005";
		String stopRow = "0015";
		List<String> columns = new ArrayList<String>(0);
		columns.add("info:birthday");
		int id = client.scannerOpen(table, startRow, stopRow, columns, attributes);
		int nbRows = 2;
		List<TRowResult> results = client.scannerGetList(id, nbRows);
		while(results != null && !results.isEmpty()) {
			for(TRowResult result : results) {
				client.iterateResults(result);
			}
			results = client.scannerGetList(id, nbRows);
		}
		client.scannerClose(id);
	}
	
	public void caseForGet() throws TException {
		Map<String, String> attributes = new HashMap<String, String>(0);
		String table = setTable();
		List<String> rows = new ArrayList<String>(0);
		rows.add("0009");
		rows.add("0098");
		rows.add("0999");
		List<String> columns = new ArrayList<String>(0);
		columns.add("info:birthday");
		columns.add("info:gender");
		List<TRowResult> results = client.getRowsWithColumns(table, rows, columns, attributes);
		for(TRowResult result : results) {
			client.iterateResults(result);
		}
	}
	
	public static void main(String[] args) 
			throws IOError, IllegalArgument, TException, UnsupportedEncodingException {
		Test test = new Test();
//		test.caseForUpdate(); // insert or update rows/cells
//		test.caseForDelete(); // delete cells
//		test.caseForDeleteRow(); // delete rows
//		test.caseForScan(); // scan rows
		test.caseForGet(); // get rows
	}
	
}

上面的测试可以实现操作Hbase表数据。另外，在生成的Thrift客户端代码中，org.apache.hadoop.hbase.thrift.generated.Hbase.Iface中给出了全部的服务接口，可以根据需要来选择，客户端org.apache.hadoop.hbase.thrift.generated.Hbase.Client实现了与Thrift交互的一些逻辑的处理，通过该类对象可以代理HBase提供的Thrift服务。

参考链接

本文基于署名-非商业性使用-相同方式共享 4.0许可协议发布，欢迎转载、使用、重新发布，但务必保留文章署名时延军（包含链接：http://shiyanjun.cn），不得用于商业目的，基于本文修改后的作品务必以相同的许可发布。如有任何疑问，请与我联系。

评论(3)： “HBase Thrift客户端Java API实践”

发表评论取消回复

丘比特: 请问博主，如果在窗口中用到广播状态，现在您有什么实现方案吗？
z: 寫的真好
方俊: 大佬好有耐心，从14年回复到19年哈哈
Yanjun: 图是用 Astah 和 OminiGraffle 画的
JacobZheng: 问个题外话，图是用什么工具画的啊
Derek Dekker: 感觉还挺难的
luosijie: 博主你好，请问您知道K距离方法出自哪篇文献吗，我该如何引用？
lee: 你好，点赞
郑威: 顺便咨询一下博主，我考虑在processfunction中注册定时器，来替代timewindow；connect上...
郑威: 非常感谢，百度了好久，只有您的文章，可以完全帮助我。特意评论感谢
rentao: 我测试过在Drvier端创建pool对象然后进行广播，但在e xcutor中通过broadcast.value获取的...
rentao: 很好的文章。你在第一个例子中使用单例对象来定义redisPo ol，而scala中单例对象是具有懒加载属性的，我理解...

评论(3)： “HBase Thrift客户端Java API实践”

发表评论 取消回复

发表评论取消回复