HbaseUtils，HBase Java Api

 

 

 

package com.xx.xx.service.spark;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.FilterList.Operator;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.log4j.Logger;

import com.xx.xx.common.util.PropertiesUtils;

public class HbaseUtils {
	private final static Logger LOG = Logger.getLogger(HbaseUtils.class);
	private static final String zookeeperQuorum = (String) PropertiesUtils
			.getJddpConf().get("zookeeperQuorum");
	private static HTablePool tablePool;
	private final static int BATCH_SIZE = 1000;
	public final static String QUERY_REPORT_TABLE = "query_report";
	public final static String QUERY_STATUS_TABLE = "query_status";
	public final static String CONSOLE_LOG_TABLE = "console_log";
	public final static String TRY_RUN_RESULT_TABLE = "try_run_result";
	public final static String COLUMN_FAMILY = "cf";
	public final static String QUALIFIER = "c";

	static {
		Configuration conf = new Configuration();
		LOG.info("zookeeperQuorum: " + zookeeperQuorum);
		conf.set("hbase.zookeeper.quorum", zookeeperQuorum);
		Configuration hbaseConf = HBaseConfiguration.create(conf);
		tablePool = new HTablePool(hbaseConf, 30);
	}

	private List<Result> getResult(String hbaseTableName, String columnFamily,
			String qualifier, String keyPrefix) {
		List<Result> result = new ArrayList<Result>();
		HTableInterface table = null;
		try {
			table = tablePool.getTable(hbaseTableName);
			Scan scan = new Scan();
			List<Filter> filters = new ArrayList<Filter>();
			Filter prifixFilter = new PrefixFilter(keyPrefix.getBytes());
			filters.add(prifixFilter);
			Filter allFilter = new FilterList(Operator.MUST_PASS_ALL, filters);
			scan.setFilter(allFilter);

			ResultScanner rs = table.getScanner(scan);
			try {
				for (Result r : rs) {
					result.add(r);
				}
			} finally {
				rs.close();
			}
		} catch (Throwable e) {
			LOG.error("ERROR: get table: " + hbaseTableName + ", prefix key:"
					+ keyPrefix, e);
		} finally {
			if (table != null)
				try {
					table.close();
				} catch (IOException e) {
					LOG.error("close table error, get table: " + hbaseTableName
							+ ", prefix key:" + keyPrefix, e);
				}
		}
		return result;
	}

	private Map<String, String> read(String hbaseTableName, String columnFamily,
			String qualifier, String keyPrefix) {
		Map<String, String> result = new HashMap<String, String>();
		List<Result> resultList = getResult(hbaseTableName, columnFamily,
				qualifier, keyPrefix);
		for (Result r : resultList) {
			KeyValue[] kv = r.raw();
			for (int i = 0; i < kv.length; i++) {
				result.put(new String(kv[i].getRow()), new String(kv[i].getValue()));
			}
		}
		return result;
	}

	private void write(String hbaseTableName, String columnFamily,
			String qualifier, String keyPrefix, Collection<String> contents) {
		HTableInterface table = null;
		try {
			table = tablePool.getTable(hbaseTableName);
			List<Put> putList = new ArrayList<Put>();
			int idx = 0;
			for (String line : contents) {
				String rowKey = keyPrefix + idx;
				if (contents.size() == 1)
					rowKey = keyPrefix;
				idx++;
				Put put = new Put(rowKey.getBytes());
				put.add(columnFamily.getBytes(), qualifier.getBytes(),
						line.getBytes());
				putList.add(put);
				if (putList.size() >= BATCH_SIZE) {
					table.put(putList);
					table.flushCommits();
					putList.clear();
				}
			}
			table.put(putList);
			table.flushCommits();
		} catch (Throwable e) {
			LOG.error("ERROR: write into table: " + hbaseTableName + ", prefix key:"
					+ keyPrefix, e);
		} finally {
			if (table != null) {
				try {
					table.close();
				} catch (IOException e) {
					LOG.error("close table error, write into table: "
							+ hbaseTableName + ", prefix key:" + keyPrefix, e);
				}
			}
		}
	}

	private void delete(String hbaseTableName, String columnFamily,
			String qualifier, Collection<String> rowKeys) {
		HTableInterface table = null;
		try {
			table = tablePool.getTable(hbaseTableName);
			List<Delete> deleteList = new ArrayList<Delete>();
			int idx = 0;
			for (String r : rowKeys) {
				Delete del = new Delete(r.getBytes());
				deleteList.add(del);
				if (deleteList.size() >= BATCH_SIZE) {
					table.delete(deleteList);
					table.flushCommits();
					deleteList.clear();
				}
				idx++;
			}
			table.delete(deleteList);
			table.flushCommits();
			LOG.info("deleted " + idx + " rows from HBase table. "
					+ hbaseTableName);
		} catch (Throwable e) {
			LOG.error("delete from table: " + hbaseTableName, e);
		} finally {
			if (table != null) {
				try {
					table.close();
				} catch (IOException e) {
					LOG.error("close table error, delete from table: "
							+ hbaseTableName, e);
				}
			}
		}
	}
	
	public void writeQueryReport(String queryId, String reportJson) {
		String keyPrefix = queryId;
		write(QUERY_REPORT_TABLE, COLUMN_FAMILY, QUALIFIER, keyPrefix,
				Arrays.asList(reportJson));
	}

	private String getFirst(Collection<String> list) {
		for (String s : list) {
			return s;
		}
		return null;
	}
	
	public String getQueryReport(String queryId) {
		String keyPrefix = queryId;
		Collection<String> data = read(QUERY_REPORT_TABLE, COLUMN_FAMILY, QUALIFIER,
				keyPrefix).values();
		if (data == null || data.size() == 0)
			return null;
		else
			return getFirst(data);
	}

	public String getAsyncQueryStatus(String queryId) {
		String keyPrefix = queryId;
		Collection<String> data = read(QUERY_STATUS_TABLE, COLUMN_FAMILY, QUALIFIER,
				keyPrefix).values();
		if (data == null || data.size() == 0)
			return null;
		else
			return getFirst(data);
	}

	public void setAsyncQueryStatus(String queryId, String status) {
		String keyPrefix = queryId;
		write(QUERY_STATUS_TABLE, COLUMN_FAMILY, QUALIFIER, keyPrefix,
				Arrays.asList(status));
	}

	public Collection<String> getConsoleLog(String shellJobId) {
		String keyPrefix = "tryRun_" + shellJobId;
		Map<String,String> map = read(CONSOLE_LOG_TABLE, COLUMN_FAMILY, QUALIFIER, keyPrefix);
		delete(CONSOLE_LOG_TABLE, COLUMN_FAMILY, QUALIFIER, map.keySet());
		return map.values();
	}

	public Collection<String> getQueryResult(String queryId) {
		String keyPrefix = queryId;
		Map<String,String> map = read(TRY_RUN_RESULT_TABLE, COLUMN_FAMILY, QUALIFIER, keyPrefix);
		delete(TRY_RUN_RESULT_TABLE, COLUMN_FAMILY, QUALIFIER, map.keySet());
		return map.values();
	}

}