上一篇cm5.4和cdh5.4安装(http://my.oschina.net/penngo/blog/517223)，本文使用HBase的java客户端api操作Hbase。
需要用到的包可以在/opt/cloudera/parcels/CDH-5.4.7-1.cdh5.4.7.p0.3/jars找到，hbase版本1.0.0
HbaseTest3.java代码例子
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.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.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
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.client.Table;
import org.apache.hadoop.hbase.filter.BinaryPrefixComparator;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.filter.ValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
public class HbaseTest3 {
private Connection connection = null;
public HbaseTest3() {
}
public void createTable(String tableName) {
try {
System.out.println("start create table ......");
Connection con = this.getConnection();
HBaseAdmin hBaseAdmin = (HBaseAdmin) con.getAdmin();
if (hBaseAdmin.tableExists(tableName)) {// 如果存在要创建的表，那么先删除，再创建
hBaseAdmin.disableTable(tableName);
hBaseAdmin.deleteTable(tableName);
System.out.println(tableName + " is exist,detele....");
}
HTableDescriptor tableDescriptor = new HTableDescriptor(
TableName.valueOf(tableName));
tableDescriptor.addFamily(new HColumnDescriptor("column1"));
tableDescriptor.addFamily(new HColumnDescriptor("column2"));
tableDescriptor.addFamily(new HColumnDescriptor("column3"));
hBaseAdmin.createTable(tableDescriptor);
System.out.println("create table success......");
} catch (MasterNotRunningException e) {
e.printStackTrace();
} catch (ZooKeeperConnectionException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
}
public void insertData(String tableName) {
System.out.println("start insert data ......");
try {
Connection con = this.getConnection();
Table table = con.getTable(TableName.valueOf(tableName));
// HTablePool pool = new HTablePool(configuration, 1000);
// HTable table = (HTable) pool.getTable(tableName);
Put put = new Put("112233bbbcccc".getBytes());// 一个PUT代表一行数据，再NEW一个PUT表示第二行数据,每行一个唯一的ROWKEY，此处rowkey为put构造方法中传入的值
put.addColumn("column1".getBytes(), null, "aaa2".getBytes());// 本行数据的第一列
put.addColumn("column2".getBytes(), null, "bbb2".getBytes());// 本行数据的第三列
put.addColumn("column3".getBytes(), null, "ccc2".getBytes());// 本行数据的第三列
table.put(put);
// pool.getTable(tableName).put(put);
} catch (IOException e) {
e.printStackTrace();
}
System.out.println("end insert data ......");
}
public void queryAll(String tableName) {
// HTablePool pool = new HTablePool(configuration, 1000);
// HTable table = (HTable) pool.getTable(tableName);
try {
Connection con = this.getConnection();
Table table = con.getTable(TableName.valueOf(tableName));
ResultScanner rs = table.getScanner(new Scan());
for (Result r : rs) {
System.out.println("获得到rowkey:" + new String(r.getRow()));
for (KeyValue keyValue : r.raw()) {
System.out.println("列：" + new String(keyValue.getFamily())
+ "====值:" + new String(keyValue.getValue()));
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
public void queryByCondition1(String tableName) {
try {
Connection con = this.getConnection();
Table table = con.getTable(TableName.valueOf(tableName));
Get scan = new Get("112233bbbcccc".getBytes());// 根据rowkey查询
Result r = table.get(scan);
System.out.println("获得到rowkey:" + new String(r.getRow()));
for (KeyValue keyValue : r.raw()) {
System.out.println("列：" + new String(keyValue.getFamily())
+ "====值:" + new String(keyValue.getValue()));
}
} catch (IOException e) {
e.printStackTrace();
}
}
public void queryByCondition2(String tableName) {
try {
Connection con = this.getConnection();
Table table = con.getTable(TableName.valueOf(tableName));
Filter filter = new SingleColumnValueFilter(
Bytes.toBytes("column1"), null, CompareOp.EQUAL,
Bytes.toBytes("aaa2")); // 当列column1的值为aaa时进行查询
Scan s = new Scan();
s.setFilter(filter);
ResultScanner rs = table.getScanner(s);
for (Result r : rs) {
System.out.println("获得到rowkey:" + new String(r.getRow()));
for (KeyValue keyValue : r.raw()) {
System.out.println("列：" + new String(keyValue.getFamily())
+ "====值:" + new String(keyValue.getValue()));
}
}
} catch (Exception e) {
e.printStackTrace();
}
}
public void queryByCondition3(String tableName) {
try {
Connection con = this.getConnection();
Table table = con.getTable(TableName.valueOf(tableName));
// 提取rowkey以cccc结尾数据
// Filter filter1 = new RowFilter(CompareOp.EQUAL,new RegexStringComparator(".*cccc$"));
//
// 提取rowkey以包含bbb的数据
// Filter filter1 = new RowFilter(CompareOp.EQUAL,new SubstringComparator("bbb"));
// 提取rowkey以123开头的数据
Filter filter1 = new RowFilter(CompareOp.EQUAL,new BinaryPrefixComparator("1122".getBytes()));
// Filter filter1 = new RowFilter(CompareOp.EQUAL,
// new BinaryComparator(Bytes.toBytes(rowkey)));
Scan scan = new Scan();
scan.setFilter(filter1);
ResultScanner rs = table.getScanner(scan);
for (Result r : rs) {
System.out.println("获得到rowkey:" + new String(r.getRow()));
for (Cell cell : r.rawCells()) {
System.out.println("列："
+ new String(CellUtil.cloneFamily(cell)) + "====值:"
+ new String(CellUtil.cloneValue(cell)));
}
}
rs.close();
} catch (Exception e) {
e.printStackTrace();
}
}
public void queryByCondition4(String tableName) {
try {
Connection con = this.getConnection();
Table table = con.getTable(TableName.valueOf(tableName));
Filter filter1 = new ValueFilter(CompareOp.EQUAL,
new SubstringComparator("aaa2"));
Scan scan = new Scan();
scan.setFilter(filter1);
ResultScanner rs = table.getScanner(scan);
for (Result r : rs) {
System.out.println("获得到rowkey:" + new String(r.getRow()));
for (Cell cell : r.rawCells()) {
System.out.println("列："
+ new String(CellUtil.cloneFamily(cell)) + "====值:"
+ new String(CellUtil.cloneValue(cell)));
}
}
rs.close();
} catch (Exception e) {
e.printStackTrace();
}
}
public void close() throws IOException {
if (connection != null) {
connection.close();
}
}
public Connection getConnection() throws IOException {
Configuration configuration = HBaseConfiguration.create();
configuration.set("hbase.zookeeper.property.clientPort", "2181");
configuration.set("hbase.zookeeper.quorum", "192.168.17.108");
if (connection == null) {
connection = ConnectionFactory.createConnection(configuration);
}
return connection;
}
public static void main(String[] args) {
HbaseTest3 hbaseTest = new HbaseTest3();
try {
String tableName = "test1";
hbaseTest.createTable("test1");
hbaseTest.insertData("test1");
hbaseTest.queryAll("test1");
// hbaseTest.queryByCondition1(tableName);
// hbaseTest.queryByCondition2(tableName);
// hbaseTest.queryByCondition3(tableName);
// hbaseTest.queryByCondition4(tableName);
hbaseTest.close();
} catch (Exception e) {
e.printStackTrace();
}
}
}
运行结果：
时间： 2024-10-11 05:21:27