Hbase API增删改查表(包含过滤)代码,Java代码库,德仔网

代码语言
代码分类
【Java】 Hbase API增删改查表(包含过滤)
作者:伊菲 / 发布于2018/4/8/ 721
一、简单模式的实现的代码如下：

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
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.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.mapreduce.TableInputFormat;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.security.UserGroupInformation;

/**
 * Document:本类作用---->使用Hbase传统api创建hbase表、查询数据等（测试环境）
 * User: yangjf
 * Date: 2016/8/18  18:31
 * 本地模式
 * 测试通过
 */

public class SimpleHBase {

 public static void main(String[] args)throws Exception {

  System.setProperty("java.security.krb5.conf","F:/krb5.conf");//windows下 ,本地执行需要
  String tableName="greatgas:student";

  Configuration configuration = HBaseConfiguration.create();//创建Hbase连接配置
  configuration.set("hbase.zookeeper.quorum", "slave-31.dev.cluster.enn.cn:2181");
  configuration.set("hbase.rootdir", "hdfs://mycluster/hbase");
  configuration.set("hadoop.security.authentication", "kerberos");
  configuration.set("hbase.security.authentication", "kerberos");
  configuration.set("hbase.security.authorization", "true");
  configuration.set("hbase.master.kerberos.principal", "hbase/_HOST@ENN.CN");
  configuration.set("hbase.thrift.kerberos.principal", "hbase/_HOST@ENN.CN");
  configuration.set("hbase.regionserver.kerberos.principal", "hbase/_HOST@ENN.CN");
  configuration.set("hbase.zookeeper.property.clientPort", "2181");
  configuration.set(TableInputFormat.INPUT_TABLE, tableName);
  configuration.set("hbase.rpc.timeout", "10000");
  configuration.set("hbase.client.retries.number", "5");
  configuration.set("hbase.client.pause", "5000");
  configuration.set("hbase.client.scanner.timeout.period", "50000");
  String user = "e_heyutao@ENN.CN";
  String keyPath = "F:/e_heyutao.keytab";                 //本地测试需要的文件
  UserGroupInformation.setConfiguration(configuration);
  UserGroupInformation.loginUserFromKeytab(user, keyPath);

  //测试创建、添加、获取和删除数据
  System.out.println("建表.......");
  createTable(configuration, tableName);
  System.out.println("添加数据.......");
  addData(configuration, tableName);
  System.out.println("获取数据.......");
  getData(configuration, tableName);
  System.out.println("获取所有数据.......");
  getAllData(configuration, tableName);
  System.out.println("删除数据.......");
  deleteDate(configuration, tableName);
  System.out.println("删除表.......");
  dropTable(configuration, tableName);
  
 }
 
 /**
  * create a new Table
  * @param configuration Configuration
  * @param tableName String,the new Table's name
  * */
 public static void createTable(Configuration configuration,String tableName){
  HBaseAdmin admin;
  try {
   admin = new HBaseAdmin(configuration);
   if(admin.tableExists(tableName)){
    admin.disableTable(tableName);
    admin.deleteTable(tableName);
    System.out.println(tableName+"is exist ,delete ......");
   }
   
   
   HTableDescriptor tableDescriptor=new HTableDescriptor(TableName.valueOf(tableName));
   tableDescriptor.addFamily(new HColumnDescriptor("info"));
   tableDescriptor.addFamily(new HColumnDescriptor("address"));
   admin.createTable(tableDescriptor);
   System.out.println("end create table");
  } catch (MasterNotRunningException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  } catch (ZooKeeperConnectionException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  } catch (IOException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
  
 }
 
 /**
  * Delete the existing table
  * @param configuration Configuration
  * @param tableName String,Table's name
  * */
 public static void dropTable(Configuration configuration,String tableName){
  HBaseAdmin admin;
  try {
   admin = new HBaseAdmin(configuration);
   if(admin.tableExists(tableName)){
    admin.disableTable(tableName);
    admin.deleteTable(tableName);
    System.out.println(tableName+"delete success!");
   }else{
    System.out.println(tableName+"Table does not exist!");
   }
  } catch (MasterNotRunningException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  } catch (ZooKeeperConnectionException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  } catch (IOException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
 }
 
 /**
  * insert a data
  * @param configuration Configuration
  * @param tableName String,Table's name
  * */
 public static void addData(Configuration configuration,String tableName){
  HBaseAdmin admin;
  try {
   admin = new HBaseAdmin(configuration);
   if(admin.tableExists(tableName)){
    HTable table=new HTable(configuration, tableName);
    Put put=new Put(Bytes.toBytes("zhangsan"));
    put.add(Bytes.toBytes("info"), Bytes.toBytes("age"), Bytes.toBytes("28"));
    table.put(put);
    System.out.println("add success!");
   }else{
    System.out.println(tableName+"Table does not exist!");
   }
  } catch (MasterNotRunningException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  } catch (ZooKeeperConnectionException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  } catch (IOException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
 }
 
 /**
  * Delete a data
  * @param configuration Configuration
  * @param tableName String,Table's name
  * */
 public static void deleteDate(Configuration configuration,String tableName){
  HBaseAdmin admin;
  try {
   admin=new HBaseAdmin(configuration);
   if(admin.tableExists(tableName)){
    HTable table=new HTable(configuration, tableName);
    Delete delete=new Delete(Bytes.toBytes("zhangsan"));
    table.delete(delete);
    System.out.println("delete success!");
   }else{
    System.out.println("Table does not exist!");
   }
  } catch (MasterNotRunningException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  } catch (ZooKeeperConnectionException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  } catch (IOException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
 }
 

 /**
  * get a data
  * @param configuration Configuration
  * @param tableName String,Table's name
  * */
 public static void getData(Configuration configuration,String tableName){
  HTable table;
  try {
   table = new HTable(configuration, tableName);
   Get get=new Get(Bytes.toBytes("zhangsan"));
   Result result=table.get(get);
 
   for(Cell cell:result.rawCells()){   
    System.out.println("RowName:"+new String(CellUtil.cloneRow(cell))+" ");
    System.out.println("Timetamp:"+cell.getTimestamp()+" ");
    System.out.println("column Family:"+new String(CellUtil.cloneFamily(cell))+" ");
    System.out.println("row Name:"+new String(CellUtil.cloneQualifier(cell))+" ");
    System.out.println("value:"+new String(CellUtil.cloneValue(cell))+" ");
   }
  } catch (IOException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
 }
 
 /**
  * insert all data
  * @param configuration Configuration
  * @param tableName String,Table's name
  * */
 public static void getAllData(Configuration configuration,String tableName){
  HTable table;
  try {
   table=new HTable(configuration, tableName);
   Scan scan=new Scan();
   ResultScanner results=table.getScanner(scan);
   for(Result result:results){
    for(Cell cell:result.rawCells()){   
     System.out.println("RowName:"+new String(CellUtil.cloneRow(cell))+" ");
     System.out.println("Timetamp:"+cell.getTimestamp()+" ");
     System.out.println("column Family:"+new String(CellUtil.cloneFamily(cell))+" ");
     System.out.println("row Name:"+new String(CellUtil.cloneQualifier(cell))+" ");
     System.out.println("value:"+new String(CellUtil.cloneValue(cell))+" ");
    }
   }
  } catch (IOException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
    
 }
}


二、有接口类和实现类的实现代码如下：



1、接口类

package local.interfaces;

import java.util.List;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;

public interface HBaseDAO {

	public void save(Put put, String tableName) ;
	public void insert(String tableName, String rowKey, String family, String quailifer, String value) ;
	public void insert(String tableName, String rowKey, String family, String quailifer[], String value[]) ;
	
	public void save(List Put, String tableName) ;
	
	public Result getOneRow(String tableName, String rowKey) ;
	
	public List getRows(String tableName, String rowKey_like) ;
	
	public List getRows(String tableName, String rowKeyLike, String cols[]) ;
	
	public List getRows(String tableName, String startRow, String stopRow) ;
	
	public void deleteRecords(String tableName, String rowKeyLike);
	public   void deleteTable(String tableName);
	public   void createTable(String tableName, String[] columnFamilys);
}

2、实现类
package local.impl;

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

import local.interfaces.HBaseDAO;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
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.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTableInterface;
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.PrefixFilter;
import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
import org.apache.hadoop.security.UserGroupInformation;

/**
 * Document:本类作用---->接口实现类，增删改查
 * User: yangjf
 * Date: 2016/9/18  18:31
 * 本地模式
 * 测试通过
 */

public class HBaseDAOImp implements HBaseDAO {

	HConnection hTablePool = null;

	static Configuration conf =null;
	public HBaseDAOImp()
	{
		System.setProperty("java.security.krb5.conf","F:/krb5.conf");//windows下 ,本地执行需要
		conf = new Configuration();
		String zk_list = "slave-31.dev.cluster.enn.cn:2181";
		conf.set("hbase.zookeeper.quorum", zk_list);

//		conf.set("hbase.zookeeper.quorum", "slave-31.dev.cluster.enn.cn:2181");
		conf.set("hbase.rootdir", "hdfs://mycluster/hbase");
		conf.set("hadoop.security.authentication", "kerberos");
		conf.set("hbase.security.authentication", "kerberos");
		conf.set("hbase.security.authorization", "true");
		conf.set("hbase.master.kerberos.principal", "hbase/_HOST@ENN.CN");
		conf.set("hbase.thrift.kerberos.principal", "hbase/_HOST@ENN.CN");
		conf.set("hbase.regionserver.kerberos.principal", "hbase/_HOST@ENN.CN");
		conf.set("hbase.zookeeper.property.clientPort", "2181");
//		conf.set(TableInputFormat.INPUT_TABLE, tableName);
		conf.set("hbase.rpc.timeout", "10000");
		conf.set("hbase.client.retries.number", "5");
		conf.set("hbase.client.pause", "5000");
		conf.set("hbase.client.scanner.timeout.period", "50000");
		String user = "e_heyutao@ENN.CN";
		String keyPath = "F:/e_heyutao.keytab";                 //本地测试需要的文件
		UserGroupInformation.setConfiguration(conf);

		try {
			UserGroupInformation.loginUserFromKeytab(user, keyPath);

			hTablePool = HConnectionManager.createConnection(conf) ;
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	//main方法测试
	public static void main(String[] args) {
		HBaseDAO dao = new HBaseDAOImp();

//		创建表
		String tableName="greatgas:test";
		String cfs[] = {"cf"};
		dao.createTable(tableName,cfs);

//		存入一条数据
		Put put = new Put("mytest".getBytes());
		put.add("cf".getBytes(), "name".getBytes(), "cai10".getBytes()) ;
		dao.save(put, "greatgas:test") ;

//		插入多列数据
		Put puts = new Put("mytest".getBytes());
		List list = new ArrayList();
		put.add("cf".getBytes(), "addr".getBytes(), "shanghai1".getBytes()) ;
		put.add("cf".getBytes(), "age".getBytes(), "30".getBytes()) ;
		put.add("cf".getBytes(), "tel".getBytes(), "13889891818".getBytes()) ;
		list.add(puts) ;
		dao.save(list, "greatgas:test");

//		插入单行数据
		dao.insert("greatgas:test", "testrow", "cf", "age", "35") ;
		dao.insert("greatgas:test", "testrow", "cf", "cardid", "12312312335") ;
		dao.insert("greatgas:test", "testrow", "cf", "tel", "13512312345") ;

		List list2 = dao.getRows("greatgas:test", "testrow",new String[]{"age"}) ;
		for(Result rs : list2)
		{
			for(Cell cell:rs.rawCells()){
				System.out.println("RowName:"+new String(CellUtil.cloneRow(cell))+" ");
				System.out.println("Timetamp:"+cell.getTimestamp()+" ");
				System.out.println("column Family:"+new String(CellUtil.cloneFamily(cell))+" ");
				System.out.println("row Name:"+new String(CellUtil.cloneQualifier(cell))+" ");
				System.out.println("value:"+new String(CellUtil.cloneValue(cell))+" ");
			}
		}

		Result rs = dao.getOneRow("greatgas:test", "testrow");
		System.out.println(new String(rs.getValue("cf".getBytes(), "age".getBytes())));


	}
	@Override
	public void save(Put put, String tableName) {
		// TODO Auto-generated method stub
		HTableInterface table = null;
		try {
			table = hTablePool.getTable(tableName) ;
			table.put(put) ;

		} catch (Exception e) {
			e.printStackTrace() ;
		}finally{
			try {
				table.close() ;
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
	@Override
	public void insert(String tableName, String rowKey, String family,
					   String quailifer, String value) {
		// TODO Auto-generated method stub
		HTableInterface table = null;
		try {
			table = hTablePool.getTable(tableName) ;
			Put put = new Put(rowKey.getBytes());
			put.add(family.getBytes(), quailifer.getBytes(), value.getBytes()) ;
			table.put(put);
		} catch (Exception e) {
			e.printStackTrace();
		}finally
		{
			try {
				table.close() ;
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
	@Override
	public void insert(String tableName,String rowKey,String family,String quailifer[],String value[])
	{
		HTableInterface table = null;
		try {
			table = hTablePool.getTable(tableName) ;
			Put put = new Put(rowKey.getBytes());
			// 批量添加
			for (int i = 0; i < quailifer.length; i++) {
				String col = quailifer[i];
				String val = value[i];
				put.add(family.getBytes(), col.getBytes(), val.getBytes());
			}
			table.put(put);
		} catch (Exception e) {
			e.printStackTrace();
		}finally
		{
			try {
				table.close() ;
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
	@Override
	public void save(List Put, String tableName) {
		// TODO Auto-generated method stub
		HTableInterface table = null;
		try {
			table = hTablePool.getTable(tableName) ;
			table.put(Put) ;
		}
		catch (Exception e) {
			// TODO: handle exception
		}finally
		{
			try {
				table.close() ;
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

	}


	@Override
	public Result getOneRow(String tableName, String rowKey) {
		// TODO Auto-generated method stub
		HTableInterface table = null;
		Result rsResult = null;
		try {
			table = hTablePool.getTable(tableName) ;
			Get get = new Get(rowKey.getBytes()) ;
			rsResult = table.get(get) ;
		} catch (Exception e) {
			e.printStackTrace() ;
		}
		finally
		{
			try {
				table.close() ;
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return rsResult;
	}

	@Override
	public List getRows(String tableName, String rowKeyLike) {
		// TODO Auto-generated method stub
		HTableInterface table = null;
		List list = null;
		try {
			table = hTablePool.getTable(tableName) ;
			PrefixFilter filter = new PrefixFilter(rowKeyLike.getBytes());
			Scan scan = new Scan();
			scan.setFilter(filter);
			ResultScanner scanner = table.getScanner(scan) ;
			list = new ArrayList() ;
			for (Result rs : scanner) {
				list.add(rs) ;
			}
		} catch (Exception e) {
			e.printStackTrace() ;
		}
		finally
		{
			try {
				table.close() ;
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return list;
	}

	@Override
	public List getRows(String tableName, String rowKeyLike ,String cols[]) {
		// TODO Auto-generated method stub
		HTableInterface table = null;
		List list = null;
		try {
			table = hTablePool.getTable(tableName) ;
			PrefixFilter filter = new PrefixFilter(rowKeyLike.getBytes());
			Scan scan = new Scan();
			for (int i = 0; i < cols.length; i++) {
				scan.addColumn("cf".getBytes(), cols[i].getBytes()) ;
			}
			scan.setFilter(filter);
			ResultScanner scanner = table.getScanner(scan) ;
			list = new ArrayList() ;
			for (Result rs : scanner) {
				list.add(rs) ;
			}
		} catch (Exception e) {
			e.printStackTrace() ;
		}
		finally
		{
			try {
				table.close() ;
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return list;
	}
	@Override
	public List getRows(String tableName,String startRow,String stopRow)
	{
		HTableInterface table = null;
		List list = null;
		try {
			table = hTablePool.getTable(tableName) ;
			Scan scan = new Scan() ;
			scan.setStartRow(startRow.getBytes()) ;
			scan.setStopRow(stopRow.getBytes()) ;
			ResultScanner scanner = table.getScanner(scan) ;
			list = new ArrayList() ;
			for (Result rsResult : scanner) {
				list.add(rsResult) ;
			}

		}catch (Exception e) {
			e.printStackTrace() ;
		}
		finally
		{
			try {
				table.close() ;
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return list;
	}

	@Override
	public void deleteRecords(String tableName, String rowKeyLike){
		HTableInterface table = null;
		try {
			table = hTablePool.getTable(tableName) ;
			PrefixFilter filter = new PrefixFilter(rowKeyLike.getBytes());
			Scan scan = new Scan();
			scan.setFilter(filter);
			ResultScanner scanner = table.getScanner(scan) ;
			List list = new ArrayList() ;
			for (Result rs : scanner) {
				Delete del = new Delete(rs.getRow());
				list.add(del) ;
			}
			table.delete(list);
		}
		catch (Exception e) {
			e.printStackTrace() ;
		}
		finally
		{
			try {
				table.close() ;
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

	}

	public   void createTable(String tableName, String[] columnFamilys){
		try {
			// admin 对象
			HBaseAdmin admin = new HBaseAdmin(conf);
			if (admin.tableExists(tableName)) {
				System.err.println("此表，已存在！");
			} else {
				HTableDescriptor tableDesc = new HTableDescriptor(
						TableName.valueOf(tableName));

				for (String columnFamily : columnFamilys) {
					tableDesc.addFamily(new HColumnDescriptor(columnFamily));
				}

				admin.createTable(tableDesc);
				System.err.println("建表成功!");

			}
			admin.close();// 关闭释放资源
		} catch (MasterNotRunningException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (ZooKeeperConnectionException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}

	/**
	 * 删除一个表
	 *
	 * @param tableName
	 *            删除的表名
	 * */
	public   void deleteTable(String tableName)   {
		try {
			HBaseAdmin admin = new HBaseAdmin(conf);
			if (admin.tableExists(tableName)) {
				admin.disableTable(tableName);// 禁用表
				admin.deleteTable(tableName);// 删除表
				System.err.println("删除表成功!");
			} else {
				System.err.println("删除的表不存在！");
			}
			admin.close();
		} catch (MasterNotRunningException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (ZooKeeperConnectionException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	/**
	 * 查询表中所有行
	 * @param tablename
	 */
	public void scaner(String tablename) {
		try {
			HTable table =new HTable(conf, tablename);
			Scan s =new Scan();
			ResultScanner rs = table.getScanner(s);
			for (Result r : rs) {
				KeyValue[] kv = r.raw();
				for (int i =0; i < kv.length; i++) {
					System.out.print(new String(kv[i].getRow()) +"");
					System.out.print(new String(kv[i].getFamily()) +":");
					System.out.print(new String(kv[i].getQualifier()) +"");
					System.out.print(kv[i].getTimestamp() +"");
					System.out.println(new String(kv[i].getValue()));
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}


}
试试其它关键字
同语言下
可能有用的
伊菲贡献的其它代码(25)