現有以下關系型數據庫中的表(見表4-20表4-21和表4-22),要求將具轉換為適合Hbase存儲的表並插入數據。

① createTable(String tableName, String[] fields)

創建表，參數tableName為表的名稱，字符串數組fields為存儲記錄各個域名稱的數組。要 求當HBase已經存在名為tableName的表的時候，先刪除原有的表，再創建新的表。

import java.io.IOException;

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


public class F_createTable {
    public static Configuration configuration;
    public static Connection connection;
    public static Admin admin;
    
    /**
     * @param args
     * @throws IOException 
     */
    public static void main(String[] args) throws IOException {
        // TODO Auto-generated method stub
        createTable("Score",new String[]{"sname","course"});
        
    }
    //建立連接
    public static void init(){
        configuration  = HBaseConfiguration.create();
        configuration.set("hbase.rootdir","hdfs://localhost:9000/hbase");
        try{
            connection = ConnectionFactory.createConnection(configuration);
            admin = connection.getAdmin();
        }catch (IOException e){
            e.printStackTrace();
        }
    }
    //關閉連接
    public static void close(){
        try{
            if(admin != null){
                admin.close();
            }
            if(null != connection){
                connection.close();
            }
        }catch (IOException e){
            e.printStackTrace();
        }
    }
    /**
     * F_
     */
    public static void createTable(String myTableName,String[] colFamily) throws IOException {
 
        init();
        TableName tableName = TableName.valueOf(myTableName);
 
        if(admin.tableExists(tableName)){
            deleteTable(tableName.getNameAsString());
            System.out.println("talbe is exists,it will be deleted");
        }else {
            HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName);
            for(String str:colFamily){
                HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(str);
                hTableDescriptor.addFamily(hColumnDescriptor);
            }
            admin.createTable(hTableDescriptor);
            System.out.println("create table success");
        }
        close();
    }
    /**
     * F_
     */
    public static void deleteTable(String tableName) throws IOException {
        init();
        TableName tn = TableName.valueOf(tableName);
        if (admin.tableExists(tn)) {
            admin.disableTable(tn);
            admin.deleteTable(tn);
        }
        close();
    }
}

② addRecord（String tableName, String row, String[] fields, String。values）

向表tableName、行 row （用 S_Name表示）和字符串數組fields指定的單元格中添加對應的 數 據 valueso其 中 fields中每個元素如果對應的列族下還有相應的列限定符的話，用"colurnnFamilyicolumn"表示。例如，同 時 向 "Math" "Computer Science" "English” 3 列添加成 績時，字符串數組 fields 為{ "Score:Math” , “Score： Computer Science" , **Score:English" } , 數組 values存儲這3 門課的成績。

import java.io.IOException;

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 org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;


public class G_addRecord {
    public static Configuration configuration;
    public static Connection connection;
    public static Admin admin;
    
    /**
     * @param args
     * @throws IOException 
     */
    public static void main(String[] args) throws IOException {
        // TODO Auto-generated method stub
        addRecord("student", "2015003", new String[]{"info:S_age"}, "99");
        B_getAllData show = new B_getAllData();
        show.getTableData("student");
    }
    //建立連接
    public static void init(){
        configuration  = HBaseConfiguration.create();
        configuration.set("hbase.rootdir","hdfs://localhost:9000/hbase");
        try{
            connection = ConnectionFactory.createConnection(configuration);
            admin = connection.getAdmin();
        }catch (IOException e){
            e.printStackTrace();
        }
    }
    //關閉連接
    public static void close(){
        try{
            if(admin != null){
                admin.close();
            }
            if(null != connection){
                connection.close();
            }
        }catch (IOException e){
            e.printStackTrace();
        }
    }
    /**
     * G_
     */
    public static void addRecord(String tableName,String rowKey,String cols[],String val) throws IOException {
        init();
        Table table = connection.getTable(TableName.valueOf(tableName));
        for(String str:cols)
        {
            String[] cols_split = str.split(":");
            Put put = new Put(rowKey.getBytes());
            put.addColumn(cols_split[0].getBytes(), cols_split[1].getBytes(), val.getBytes());
            table.put(put);
        }
        table.close();
        close();
    }

}

③ scanColumn（String tableName, String column）

瀏覽表tableName某一列的數據，如果某一行記錄中該列數據不存在，則返回null 要求當 參數column為某一列族名稱時，如果底下有若干個列限定符，則要列出每個列限定符代表的列的 數據；當參數column為某一列具體名稱（如 "Score:Math” ）時，只需要列出該列的數據。

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.HTableDescriptor;
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 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;


public class H_scanColumn {
    public static Configuration configuration;
    public static Connection connection;
    public static Admin admin;
    
    /**
     * @param args
     * @throws IOException 
     */
    public static void main(String[] args) throws IOException {
        // TODO Auto-generated method stub
        scanColumn("SC","SC_score");
    }
    //建立連接
    public static void init(){
        configuration  = HBaseConfiguration.create();
        configuration.set("hbase.rootdir","hdfs://localhost:9000/hbase");
        try{
            connection = ConnectionFactory.createConnection(configuration);
            admin = connection.getAdmin();
        }catch (IOException e){
            e.printStackTrace();
        }
    }
    //關閉連接
    public static void close(){
        try{
            if(admin != null){
                admin.close();
            }
            if(null != connection){
                connection.close();
            }
        }catch (IOException e){
            e.printStackTrace();
        }
    }
    /*
     * H_
     */
    public static void scanColumn(String tablename,String col) throws IOException {
        init();
        HTableDescriptor hTableDescriptors[] = admin.listTables();
        String[] cols_split = col.split(":");
        System.out.println(tablename);
        Table table = connection.getTable(TableName.valueOf(tablename));
        //創建一個空的Scan實例
        Scan scan1 = new Scan();
        //在行上獲取遍歷器
        ResultScanner scanner1 = table.getScanner(scan1);
        //打印行的值
        for (Result res : scanner1) {
            Cell[] cells = res.rawCells();
            for(Cell cell:cells){
                 String colF = new String(CellUtil.cloneFamily(cell));
                 String col_son = new String(CellUtil.cloneQualifier(cell));
                 //System.out.println(colF+" "+col_son+" "+cols_split[0]+" "+cols_split[1]);
                if((colF.equals(cols_split[0])&&cols_split.length==1)||(colF.equals(cols_split[0])&&col_son.equals(cols_split[1])))
                    System.out.println(new String(CellUtil.cloneRow(cell))+" "+new String(CellUtil.cloneFamily(cell))+" "+new String(CellUtil.cloneQualifier(cell))+" "+new String(CellUtil.cloneValue(cell))+" ");
            }
        }
        //關閉釋放資源
        scanner1.close();
        table.close();
        close();
    }

}

④ modifyData（String tableName, String row, String column）。

修改表tableName,行 row（可以用學生姓名S_Naine表示）,列 column指定的單元格的數據。

import java.io.IOException;

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 org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;


public class I_modifyData {
    public static Configuration configuration;
    public static Connection connection;
    public static Admin admin;
    
    /**
     * @param args
     * @throws IOException 
     */
    public static void main(String[] args) throws IOException {
        // TODO Auto-generated method stub
        modifyData("student","2015003","info:S_name","Sunxiaochuan");
        B_getAllData show = new B_getAllData();
        show.getTableData("student");
    }
    //建立連接
    public static void init(){
        configuration  = HBaseConfiguration.create();
        configuration.set("hbase.rootdir","hdfs://localhost:9000/hbase");
        try{
            connection = ConnectionFactory.createConnection(configuration);
            admin = connection.getAdmin();
        }catch (IOException e){
            e.printStackTrace();
        }
    }
    //關閉連接
    public static void close(){
        try{
            if(admin != null){
                admin.close();
            }
            if(null != connection){
                connection.close();
            }
        }catch (IOException e){
            e.printStackTrace();
        }
    }
    //I_
    public static void modifyData(String tableName,String rowKey,String col,String val) throws IOException {
        init();
        Table table = connection.getTable(TableName.valueOf(tableName));
        Put put = new Put(rowKey.getBytes());
        String[] cols_split = col.split(":");
        if(cols_split.length==1)
        {
            cols_split = insert(cols_split,"");
        }
        put.addColumn(cols_split[0].getBytes(), cols_split[1].getBytes(), val.getBytes());
        table.put(put);
        table.close();
        close();
    }
    //I_
    private static String[] insert(String[] arr, String str) {
        int size = arr.length;  //獲取數組長度
        String[] tmp = new String[size + 1];  //新建臨時字符串數組，在原來基礎上長度加一
        for (int i = 0; i < size; i++){  //先遍歷將原來的字符串數組數據添加到臨時字符串數組
            tmp[i] = arr[i];
        }
        tmp[size] = str;  //在最后添加上需要追加的數據
        return tmp;  //返回拼接完成的字符串數組
    }
}

⑤ deleteRow（Slring tableName, String row）。

刪除表tableName中row指定的行的記錄。

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import java.io.IOException;
import java.util.Collection;
import java.util.Iterator;
 
public class J_deleteRow {
    public static Configuration configuration;
    public static Connection connection;
    public static Admin admin;
    
    /**
     * @param args
     * @throws IOException 
     */
    public static void main(String[] args) throws IOException {
        B_getAllData show = new B_getAllData();
        show.getTableData("student");
        deleteRow("student", "2015003");
        show.getTableData("student");
        // TODO Auto-generated method stub

    }
    //建立連接
    public static void init(){
        configuration  = HBaseConfiguration.create();
        configuration.set("hbase.rootdir","hdfs://localhost:9000/hbase");
        try{
            connection = ConnectionFactory.createConnection(configuration);
            admin = connection.getAdmin();
        }catch (IOException e){
            e.printStackTrace();
        }
    }
    //關閉連接
    public static void close(){
        try{
            if(admin != null){
                admin.close();
            }
            if(null != connection){
                connection.close();
            }
        }catch (IOException e){
            e.printStackTrace();
        }
    }
    /**
     * J_
     */
    public static void deleteRow(String tableName,String rowKey) throws IOException {
        init();
        Table table = connection.getTable(TableName.valueOf(tableName));
        Delete delete = new Delete(rowKey.getBytes());
        table.delete(delete);
        table.close();
        close();
    }
}