HBase开发

Admin接口
Admin接口位于org.apache.hadoop.hbase.client包

下面以一个简单的程序为例，说明Hbase Java程序的开发过程
新建Java Project

添加依赖

    4.0.0

    com.example
    myHbase
    1.0-SNAPSHOT
    
        
            org.apache.hbase
            hbase-server
            1.2.6
        
        
            org.apache.hbase
            hbase-client
            1.2.6
        
    

将hbase和hadoop配置文件放入资源目录
进入集群将配置文件夹中的hdfs-site.xml、core-site.xml、hbase-site.xml放入IDEA项目资源目录。

创建log4j.properties配置文件

# priority  :debug> Method: %l ]%n%p:%m%n
#debug log
log4j.logger.debug=debug
log4j.appender.debug=org.apache.log4j.DailyRollingFileAppender 
log4j.appender.debug.DatePattern='_'yyyy-MM-dd'.log'
log4j.appender.debug.File=./src/com/hp/log/debug.log
log4j.appender.debug.Append=true
log4j.appender.debug.Threshold=DEBUG
log4j.appender.debug.layout=org.apache.log4j.PatternLayout 
log4j.appender.debug.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss a} [Thread: %t][ Class:%c >> Method: %l ]%n%p:%m%n
#warn log
log4j.logger.warn=warn
log4j.appender.warn=org.apache.log4j.DailyRollingFileAppender 
log4j.appender.warn.DatePattern='_'yyyy-MM-dd'.log'
log4j.appender.warn.File=./src/com/hp/log/warn.log
log4j.appender.warn.Append=true
log4j.appender.warn.Threshold=WARN
log4j.appender.warn.layout=org.apache.log4j.PatternLayout 
log4j.appender.warn.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss a} [Thread: %t][ Class:%c >> Method: %l ]%n%p:%m%n
#error
log4j.logger.error=error
log4j.appender.error = org.apache.log4j.DailyRollingFileAppender
log4j.appender.error.DatePattern='_'yyyy-MM-dd'.log'
log4j.appender.error.File = ./src/com/hp/log/error.log 
log4j.appender.error.Append = true
log4j.appender.error.Threshold = ERROR 
log4j.appender.error.layout = org.apache.log4j.PatternLayout
log4j.appender.error.layout.ConversionPattern = %d{yyyy-MM-dd HH:mm:ss a} [Thread: %t][ Class:%c >> Method: %l ]%n%p:%m%n

新建class类
在src/main/java下创建com.example包，并在包下创建一个HbaseTest类。

package com.example;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HbaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;

import java.io.IOException;

public class HbaseTest {
    Connection connection=null;
    Admin admin=null;
    public Boolean init(){  //初始化代码，读取配置文件，完成连接，获取Admin对象
        Configuration conf= HbaseConfiguration.create();
        // conf.set("hbase.rootdir","hdfs://localhost:9000/hbase");
        try{
            this.connection= ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
        try {
            this.admin=this.connection.getAdmin();
        } catch (IOException e) {
            e.printStackTrace();
            try {
                this.connection.close();
            } catch (IOException e1) {
                e1.printStackTrace();
                return false;
            }
        }
        return true;
    }
    
    public void close(){
        if(admin!=null){
            try{
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        if(connection!=null){
            try{
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    
    public static void main(String[] args){
        HbaseTest test=new HbaseTest();
        if(! test.init())return;
        try{
            System.out.println(test.admin.tableExists(TableName.valueOf("student")));
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            test.close();
        }
    }
}

运行HbaseTest
运行前需要启动集群，并保证hadoop和hbase运行正常。

显示表及其列族
showAllTable.java

package com.example;

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.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;

import java.io.IOException;

public class ShowAllTable extends HbaseTest{
    public void showAllTable() throws IOException {
        HTableDescriptor []htables=admin.listTables();
        for(HTableDescriptor hdes:htables){
            HColumnDescriptor []cdeses=hdes.getColumnFamilies();
            System.out.println(" ==================== ");
            System.out.println("table:"+hdes.getNameAsString()+"tColumn Families");
            System.out.println(" ==================== ");
            for(HColumnDescriptor cdes:cdeses){
                System.out.println(cdes.getNameAsString());
            }
        }
    }

    public static void main(String []args){
        ShowAllTable showAllTable=new ShowAllTable();
        if(!showAllTable.init())return;
        try{
            showAllTable.showAllTable();
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            showAllTable.close();
        }
    }
}

表的创建和删除

package com.example;

import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;

import java.io.IOException;

public class OperationTable extends HbaseTest {
    private void createTable(String myTableName, String[] colFamily) throws IOException {
        TableName tableName=TableName.valueOf(myTableName);
        if(admin.tableExists(tableName)){
            System.out.println("table is exists");
        }else {
            HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName);
            for (String str : colFamily) {
                HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(str);
                hColumnDescriptor.setMaxVersions(5);
                hTableDescriptor.addFamily(hColumnDescriptor);
            }
            admin.createTable(hTableDescriptor);
            System.out.println("create table success!");
        }
    }
    public static void main(String []args){
        OperationTable operationTable=new OperationTable();
        String[] colFamily={"name","id","test"};
        if(!operationTable.init())return;
        try{
            operationTable.createTable("test:HbaseTest",colFamily);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

删除表

    private void deleteTable(String tableName) throws IOException {
        TableName tabName=TableName.valueOf(tableName);
        if(admin.tableExists(tabName)){
            admin.disableTable(tabName);
            admin.deleteTable(tabName);
        }
    }

增加数据
1）编写增加数据insertCell方法
参数依次为表名、行键、列族、列限定符和值。

	public void insertCell(String tableName, String rowKey, String colFamily, String col, String val) throws IOException {
        Table table=connection.getTable(TableName.valueOf(tableName));
        Put put=new Put(rowKey.getBytes());
        put.addColumn(colFamily.getBytes(),col.getBytes(),val.getBytes());
        table.put(put);
        table.close();
    }

重载insertCell
参数依次为表名、行键、列族、列限定符、时间戳和值。

    public void insertCell(String tableName, String rowKey, String colFamily, String col,long ts, String val) throws IOException {
        Table table=connection.getTable(TableName.valueOf(tableName));
        Put put=new Put(rowKey.getBytes());
        put.addColumn(colFamily.getBytes(),col.getBytes(),ts,val.getBytes());
        table.put(put);
        table.close();
    }

编写批量增加insertList方法

    public void insertList(String tableName,String[] rowKey,String colFamily,String col,String[] val) throws IOException {
        Table table=connection.getTable(TableName.valueOf(tableName));
        ArrayListlist=new ArrayList();
        for(int i=0;i 

 追加数据
 
    public void appendCell(String tableName, String rowKey, String colFamily, String col, String val) throws IOException {
        Table table=connection.getTable(TableName.valueOf(tableName));
        Append append=new Append(rowKey.getBytes());
        append.add(colFamily.getBytes(),col.getBytes(),val.getBytes());
        table.append(append);
        table.close();
    }
 

 查看数据
 1）显示一行的内容
 
    public void showCell(Result result,boolean flag){
        List cells=result.listCells();
        if(flag) System.out.println("Row Family;QualifiertTimetamptvalue");
        for(Cell cell:cells){
            System.out.println("-----------------------------------");
            System.out.printf("%-20s",new String(cell.getRowArray(),cell.getRowOffset(),cell.getRowLength()));
            System.out.printf("%s:%st",new String(cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength()));
            System.out.printf("%15st",cell.getTimestamp());
            System.out.printf("%-20sn",new String(cell.getValueArray(),cell.getValueOffset(),cell.getValueLength()));
        }
    }
    public void getData(String tableName,String rowKey,String colFamily,String col,int maxVersions) throws IOException {
        Table table=connection.getTable(TableName.valueOf(tableName));
        Get get=new Get(rowKey.getBytes());
        if(col.equals(""))
            get.addFamily(colFamily.getBytes());
        else
            get.addColumn(colFamily.getBytes(),col.getBytes());
        if(maxVersions>1){
            get.setMaxVersions(maxVersions);
        }
        Result result=table.get(get);
        showCell(result,true);
        table.close();
    }
 

 2）扫描表中的内容
 
    public void scanData(String tableName,String colFamily,String col,int maxVersion) throws IOException {
        Table table=connection.getTable(TableName.valueOf(tableName));
        Scan scan=new Scan();
        if(col.equals(""))
            scan.addFamily(colFamily.getBytes());
        else
            scan.addColumn(colFamily.getBytes(),col.getBytes());
        if(maxVersion!=0){
            scan.setMaxVersions(maxVersion);
        }
        ResultScanner rs=table.getScanner(scan);
        boolean flag=true;
        for(Result result:rs){
            showCell(result,flag);
            flag=false;
        }
        rs.close();
        table.close();
    }
 

 删除数据
 1）删除列族
 
    public void deleteFamily(String tableName,String rowKey,String colFamily,int flag,long timeStamp) throws IOException {
        Table table=connection.getTable(TableName.valueOf(tableName));
        Delete delete=new Delete(rowKey.getBytes());
        switch (flag){
            case 0://删除指定列族的所有列和版本
                delete.addFamily(colFamily.getBytes());
                break;
            case 1://删除指定时间戳的版本
                delete.addFamilyVersion(colFamily.getBytes(),timeStamp);
                break;
            case 2://删除小于等于时间戳的版本
                delete.addFamily(colFamily.getBytes(),timeStamp);
        }
        table.delete(delete);
        table.close();
    }
 
2）删除列
 
    public void deleteColumn(String tableName, String rowKey, String colFamily, String col, int flag, long timeStamp) throws IOException {
        Table table = connection.getTable(TableName.valueOf(tableName));
        Delete delete = new Delete(rowKey.getBytes());
        switch (flag){
            case 0://删除列的所有版本
                delete.addColumns(colFamily.getBytes(),col.getBytes());
                break;
            case 1://删除指定时间戳的版本
                delete.addColumn(colFamily.getBytes(),col.getBytes(),timeStamp);
                break;
            case 2://删除小于等于时间戳的版本
                delete.addColumns(colFamily.getBytes(),col.getBytes(),timeStamp);
                break;
            case 3://删除列的最新版本
                delete.addColumn(colFamily.getBytes(),col.getBytes());
        }
        table.delete(delete);
        table.close();
    }
					
										


					

						
欢迎分享，转载请注明来源：内存溢出
原文地址: http://outofmemory.cn/zaji/4002301.html