HBase系列三

关于spring-hbase中scan + filter的操作；通过HbaseTemplate + HTableInterface完成如下操作
一、测试源码
@Component
public class HbaseUtil implements InitializingBean {
private String tableName = "users";
// table bytes
private byte[] tableNameAsBytes = Bytes.toBytes("users");
// family bytes
private static byte[] CF_INFO = Bytes.toBytes("cfInfo");
// column bytes
private byte[] qUser = Bytes.toBytes("user");
private byte[] qEmail = Bytes.toBytes("email");
private byte[] qPassword = Bytes.toBytes("password");

@Resource(name = "hbaseConfiguration")
private Configuration config;

@Autowired
private HbaseTemplate hbaseTemplate;
   
// hbase administrator
private HBaseAdmin admin;
   
// initialize
public void initialize() throws IOException {

if (admin.tableExists(tableNameAsBytes)) {
if (!admin.isTableDisabled(tableNameAsBytes)) {
System.out.printf("Disabling %s\n", tableName);
admin.disableTable(tableNameAsBytes);
}
System.out.printf("Deleting %s\n", tableName);
admin.deleteTable(tableNameAsBytes);
}

HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);
HColumnDescriptor columnDescriptor = new HColumnDescriptor(CF_INFO);
tableDescriptor.addFamily(columnDescriptor);

admin.createTable(tableDescriptor);

}
   
// add data
public User save(final String userName, final String email,
final String password) {
return hbaseTemplate.execute(tableName, new TableCallback<User>() {
public User doInTable(HTableInterface table) throws Throwable {
User user = new User(userName, email, password);
Put p = new Put(Bytes.toBytes(user.getName()));
p.add(CF_INFO, qUser, Bytes.toBytes(user.getName()));
p.add(CF_INFO, qEmail, Bytes.toBytes(user.getEmail()));
p.add(CF_INFO, qPassword, Bytes.toBytes(user.getPassword()));
table.put(p);
return user;

}
});
}

// query data
public List<User> findAll() {
return hbaseTemplate.find(tableName, "cfInfo", new RowMapper<User>() {
@Override
public User mapRow(Result result, int rowNum) throws Exception {
return new User(Bytes.toString(result.getValue(CF_INFO, qUser)),
    Bytes.toString(result.getValue(CF_INFO, qEmail)),
    Bytes.toString(result.getValue(CF_INFO, qPassword)));
}
});

}
    // scanner
public void scannerData() throws IOException{
Scan scan = new Scan();
scan.addColumn(CF_INFO, qUser);
// comparate
Filter filter = new RowFilter(CompareFilter.CompareOp.LESS_OR_EQUAL,new BinaryComparator(Bytes.toBytes("userName_15")));

// regex
// filter = new RowFilter(CompareFilter.CompareOp.EQUAL,
// new RegexStringComparator(".*."));


// substring
filter = new RowFilter(CompareFilter.CompareOp.EQUAL,new SubstringComparator("_5"));
scan.setFilter(filter);
HTableInterface htable =  HbaseUtils.getHTable(tableName, config);
ResultScanner resultScanner = null;

resultScanner = htable.getScanner(scan);
for(Result result : resultScanner){
System.out.println(result);
}

resultScanner.close();

/*try{
resultScanner = htable.getScanner(scan);
for(Result result : resultScanner){
System.out.println(result);
}
}catch(IOException io){
System.out.println(io.getMessage());
}catch(Exception exp){
System.out.println(exp.toString());
}finally{
resultScanner.close();
}*/
}

// scanner column family
public void scanColumnData() throws IOException{
// Family Column
Filter filter = new FamilyFilter(CompareFilter.CompareOp.LESS_OR_EQUAL,
new BinaryComparator(Bytes.toBytes("cfInfo")));
// Scanner
Scan scan = new Scan();
scan.setFilter(filter);
//
HTableInterface htable =  HbaseUtils.getHTable(tableName, config);
ResultScanner scanner = htable.getScanner(scan);
//
for(Result result : scanner){
System.out.println(result);
}
//
Get get1 = new Get(Bytes.toBytes("userName_3"));
get1.setFilter(filter);
Result result1 = htable.get(get1);
System.out.println(" result of get() : " + result1);
//
scanner.close();
htable.close();
}

public void scannerSpecifiedColumnData() throws IOException{
//filter
Filter filter = new ValueFilter(CompareFilter.CompareOp.GREATER_OR_EQUAL,
new SubstringComparator("_4"));
//
Scan scan = new Scan();
scan.setFilter(filter);
//
HTableInterface htable =  HbaseUtils.getHTable(tableName, config);
ResultScanner scanner = htable.getScanner(scan);
//
for(Result result : scanner){
for(KeyValue kv : result.raw()){
System.out.println(" KV: " + kv + " ,Vlaue = " + Bytes.toString(kv.getValue()));
}
}
}

public void filter(boolean drop,
CompareFilter.CompareOp operator,
ByteArrayComparable comparator) throws IOException{
Filter filter = null;
if(comparator != null){
filter = new DependentColumnFilter(Bytes.toBytes("cfInfo"),
Bytes.toBytes("email"),drop,operator,comparator);
}
else{
filter = new DependentColumnFilter(Bytes.toBytes("cfInfo"),
Bytes.toBytes("email"),drop);
}
//
Scan scan =  new Scan();
scan.setFilter(filter);
HTableInterface htable =  HbaseUtils.getHTable(tableName, config);
ResultScanner scanner = htable.getScanner(scan);
for(Result result : scanner){
for(KeyValue kv : result.raw()){
System.out.println(" KV: " + kv + " ,Vlaue = " + Bytes.toString(kv.getValue()));
}
}
scanner.close();

//
Get get = new Get(Bytes.toBytes("userName_1"));
get.setFilter(filter);
Result result = htable.get(get);
for(KeyValue kv : result.raw()){
System.out.println(" KV: " + kv + " ,Vlaue = " + Bytes.toString(kv.getValue()));
}

htable.close();
}

public void singleColumnValueFilter() throws IOException{
SingleColumnValueFilter singleFilter = new SingleColumnValueFilter(
Bytes.toBytes("cfInfo"),
Bytes.toBytes("password"),
CompareFilter.CompareOp.NOT_EQUAL,
new SubstringComparator("password_0")
);
singleFilter.setFilterIfMissing(true);
//
Scan scan = new Scan();
scan.setFilter(singleFilter);
HTableInterface htable =  HbaseUtils.getHTable(tableName, config);
ResultScanner scanner = htable.getScanner(scan);
for(Result result : scanner){
for(KeyValue kv : result.raw()){
System.out.println(" KV: " + kv + " ,Vlaue = " + Bytes.toString(kv.getValue()));
}
}
scanner.close();

System.out.println(" get + filter === ");
Get get = new Get(Bytes.toBytes("userName_8"));
get.setFilter(singleFilter);
Result result = htable.get(get);
for(KeyValue kv : result.raw()){
System.out.println(" KV: " + kv + " ,Vlaue = " + Bytes.toString(kv.getValue()));
}

htable.close();
}

@Override
public void afterPropertiesSet() throws Exception {
admin = new HBaseAdmin(config);
}
}
二、调用源码
public class App {
private static final Log log = LogFactory.getLog(App.class);

public static void main(String[] args) throws IOException {
// TODO Auto-generated method stub
AbstractApplicationContext context = new ClassPathXmlApplicationContext(
"/META-INF/spring/application-context.xml", App.class);
log.info("HBase Application Running");
context.registerShutdownHook();
// create hbase util
HbaseUtil hbase = context.getBean(HbaseUtil.class);
// execute initialize (administrator)
// hbase.initialize();
// add data
/*for(int i=0; i<= 10 ; i++){
hbase.save("userName_" + i, "email_" + (i+1), "password_" + i);
}*/
// queay
/*List<User> users = hbase.findAll();
for(User user : users){
System.out.println(user);
}*/
// scan
try{
//hbase.scannerData();
//hbase.scanColumnData();
//
//hbase.scannerSpecifiedColumnData();

// DependentColumnFilter
/*System.out.println(" no compare ...");
hbase.filter(true, CompareFilter.CompareOp.NO_OP, null);
System.out.println(" no compare ... filter...");
hbase.filter(false, CompareFilter.CompareOp.NO_OP, null);

System.out.println(" CompareFilter EQUAL ...");
hbase.filter(true, CompareFilter.CompareOp.EQUAL,
new BinaryPrefixComparator(Bytes.toBytes("email_4")));
System.out.println(" CompareFilter EQUAL ... filter...");
hbase.filter(false, CompareFilter.CompareOp.EQUAL,
new BinaryPrefixComparator(Bytes.toBytes("email_4")));

System.out.println(" RegexStringComparator EQUAL ...");
hbase.filter(true, CompareFilter.CompareOp.EQUAL,
new RegexStringComparator(".*_*"));
System.out.println(" RegexStringComparator EQUAL ... filter...");
hbase.filter(false, CompareFilter.CompareOp.EQUAL,
new RegexStringComparator(".*_*"));*/
hbase.singleColumnValueFilter();
}
catch(IOException io){
System.out.println(io.getMessage());
}
System.out.println(" hbase operation is over .......");
}

}
关于scan和filter的详细解释则可以参考《hbase权威指南》