三、JAVA调用Hbase代码：

package net.chnbs.delta.dzfp.app.common.impl;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
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.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;


import net.chnbs.delta.dzfp.app.common.CommonHadoop;


/**
* @Description TODO(分布式存储Hadoop 接口实现层)
* @author 彭甜
* @version 1.0.0 2013年7月15日 09:37:26
*/
public class CommonHadoopImp implements CommonHadoop {
protected static Logger log = Logger.getLogger(CommonHadoopImp.class);
public static Configuration configuration=null;
private static HTablePool pool =null;
public static final String ROW_KEY_VALUE="ROWKEY";//Hbase表中的行名称健
static {
configuration = HBaseConfiguration.create();
/**
* 单机调用
*/
configuration.set("hbase.zookeeper.property.clientPort", "3351");
configuration.set("hbase.zookeeper.quorum", "172.16.23.21");//设置zookeeper的主机，建议使用单数
configuration.set("hbase.master", "172.16.23.21:60000");//设置hbase的master主机名和端口


/* //集群分布式调用
configuration.set("hbase.zookeeper.property.clientPort", "2222");
configuration.set("hbase.zookeeper.quorum", "192.9.205.150");//设置zookeeper的主机，建议使用单数
configuration.set("hbase.master", "app1.wlfp.com:60000");//设置hbase的master主机名和端口
*/

pool=new HTablePool(configuration, 1000);
/*app1.wlfp.com 192.9.205.150
db1.wlfp.com 192.9.205.152
db2.wlfp.com 192.9.205.153*/
}


/**
* @Description TODO(创建没有列的表)
* @param tableName,String 表名
* @throws Exception
*/
public void createTable(String tableName) throws Exception {
// TODO Auto-generated method stub
createTable(tableName,null);
}


/**
* @Description TODO(删除指定Hbase表中的rowKey数据)
* @param tableName,String 表名
* @param hadoopMapTypeRowKey,String Hbase tableName中的 行名称
* @throws Exception
*/
public void deleteHbaseDataByRowKeyAndTableName(String tableName,
String... hadoopMapTypeRowKeys) throws Exception {
HTable table = new HTable(configuration, tableName);
log.info("开始删除"+tableName+"中的值");
//存储要删除的hadoopMapTypeRowKeys
List list = new ArrayList();
if(hadoopMapTypeRowKeys!=null){
for(String hadoopMapTypeRowKey:hadoopMapTypeRowKeys){
Delete delete = new Delete(hadoopMapTypeRowKey.getBytes());
list.add(delete);
}
}

//执行删除hadoopMapTypeRowKeys
table.delete(list);

log.info("删除行成功!");



}





@Override
public void dropTable(String tableName) throws Exception {
log.info("开始删除表"+tableName+"............");
HBaseAdmin admin = new HBaseAdmin(configuration);
admin.disableTable(tableName);
admin.deleteTable(tableName);
log.info("删除表"+tableName+"成功............");


}






/**
* @Description TODO(向指定Hbase 表中插入数据)
* @param tableName,String 表名
* @param hadoopMapTypeRowKey,String Hbase tableName中的 行名称
* @param map,Map 存储的数据
* @throws Exception
*/
public void insertHbaseData(String tableName, Map map,String hadoopMapTypeRowKey) throws Exception {

log.info("insertHbaseData start.............");

HTableInterface table = pool.getTable(tableName);
Put put = new Put(hadoopMapTypeRowKey.getBytes());// 一个PUT代表一行数据，再NEW一个PUT表示第二行数据,每行一个唯一的ROWKEY，此处rowkey为put构造方法中传入的值

/*
* 开始插入数据
*/
if(map!=null){
Set keys=map.entrySet();
if(keys!=null){
Iterator iterator = keys.iterator();
while(iterator.hasNext()){
Map.Entry<String, Object> entry=(Entry)iterator.next();
String columnKey=entry.getKey();
String columnValue=(String) entry.getValue();
//log.info("columnKey:"+columnKey+",columnValueo:"+columnValue);
put.add(columnKey.getBytes(), null, (columnValue==null"":columnValue).getBytes());//向表中制定列中插入数据columnValue.getBytes()
}
}
}

/*
* 执行插入数据
*/
table.put(put);

log.info("insertHbaseData end......");


}



/**
* @Description TODO(获取指定分布式存储的信息)
* @param tableName,String 表名
* @param hadoopMapTypeRowKey,String Hbase tableName中的 行名称
* @return Map<String,Object>
* @throws Exception
*/
public Map<String, Object> getHbaseData(String tableName,
String hadoopMapTypeRowKey) throws Exception {
//返回的Map对象
Map<String,Object> returnMap=new HashMap<String,Object>();

log.info("获取"+tableName+"中的"+hadoopMapTypeRowKey+"值......");
HTableInterface table = pool.getTable(tableName);

Get scan = new Get(hadoopMapTypeRowKey.getBytes());// 根据hadoopMapTypeRowKey查询
Result r = table.get(scan);

for (KeyValue keyValue : r.raw()) {
//log.info("列：" + new String(keyValue.getFamily())+ "====值:" + new String(keyValue.getValue()));
returnMap.put(new String(keyValue.getFamily()), new String(keyValue.getValue()));
}

return returnMap;
}


/**
* @Description TODO(查询指定Hbase中表中的数据)
* @param tableName,String 表名
* @return List
* @throws Exception
*/
public List queryAllHbaseData(String tableName) throws Exception {
log.info("查询"+tableName+"中所有数据...........");
List<Map<String,Object>> returnListMap=new ArrayList<Map<String,Object>>();//查询所返回的值信息

HTableInterface table = pool.getTable(tableName);

//执行查询指定表中所有的信息
ResultScanner rs = table.getScanner(new Scan());
for (Result r : rs) {
Map<String,Object> returnMap=new HashMap<String,Object>();
returnMap.put(ROW_KEY_VALUE, new String(r.getRow()));
//log.info("获得到rowkey:" + new String(r.getRow()));
for (KeyValue keyValue : r.raw()) {
/*log.info("列：" + new String(keyValue.getFamily())
+ "====值:" + new String(keyValue.getValue()));*/
returnMap.put(new String(keyValue.getFamily()), new String(keyValue.getValue()));
}
returnListMap.add(returnMap);
}

return returnListMap;
}


/**
* @Description TODO(根据 查询条件 查询指定HBase表中的数据 )
* @param tableName,String 表名
* @param mapParam,Map 查询条件参数
* @return List
* @throws Exception
*/
public List queryHbaseDataList(String tableName, Map mapParam)
throws Exception {
log.info("=========查询指定HBase表中"+tableName+"的数据");
List<Map<String,Object>> returnListMap=new ArrayList<Map<String,Object>>();//查询所返回的值信息

HTableInterface table = pool.getTable(tableName);


List<Filter> filters = new ArrayList<Filter>();

/*
* 开始遍历查询的条件
*/
if(mapParam!=null){
log.info("============查询的条件:"+mapParam.toString());
Set keys=mapParam.entrySet();
if(keys!=null){
Iterator iterator = keys.iterator();
while(iterator.hasNext()){
Map.Entry<String, Object> entry=(Entry)iterator.next();
String columnKey=entry.getKey();//列名
String columnValue=(String) entry.getValue();//列名多对应的值
Filter filter = new SingleColumnValueFilter(Bytes
.toBytes(columnKey), null, CompareOp.EQUAL, Bytes
.toBytes(columnValue));
filters.add(filter);
}
}
}

/**
* 执行多条件过滤查询
*/
FilterList filterList = new FilterList(filters);
Scan scan = new Scan();
scan.setFilter(filterList);

ResultScanner rs = table.getScanner(scan);
for (Result r : rs) {
Map<String,Object> returnMap=new HashMap<String,Object>();
returnMap.put(ROW_KEY_VALUE, new String(r.getRow()));
//log.info("获得到rowkey:" + new String(r.getRow()));
for (KeyValue keyValue : r.raw()) {
String columnKey=new String(keyValue.getFamily());//列名
String columnValue=new String(keyValue.getValue());//列名多对应的值
//log.info("列：" + columnKey+ "====值:" + columnValue);
returnMap.put(columnKey, columnValue);
}
returnListMap.add(returnMap);
}
rs.close();

return returnListMap;
}


/**
* @Description TODO(在HBase上创建表)
* @param tableName,String 表名
* @param columns,String[] 列名数组
* @throws Exception
*/
public void createTable(String tableName, String... columns)
throws Exception {
log.info("开始创建表名"+tableName+" ......");

if(configuration==null){
log.info("configuration is Null................");
}else{
HBaseAdmin hBaseAdmin = new HBaseAdmin(configuration);

// 如果存在要创建的表，那么先删除，再创建
if (hBaseAdmin.tableExists(tableName)) {
hBaseAdmin.disableTable(tableName);
hBaseAdmin.deleteTable(tableName);
log.info(tableName + " is exist,detele....");
}
HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);

//设置表中的列
if(columns!=null&&columns.length>0){
for(String columnName:columns){
tableDescriptor.addFamily(new HColumnDescriptor(columnName));
}

}

//创建表
hBaseAdmin.createTable(tableDescriptor);
}

log.info("创建表名"+tableName+"结束......");

}






}