HBase的訪問方式

這里只介紹三種最常用的方式

1.HBase shell

HBase的命令行工具是最簡單的接口，主要用於HBase管理

首先啟動HBase

 

 

幫助

hbase(main):001:0> help

 

 查看HBase服務器狀態

hbase(main):001:0> status

 

 

查詢HBse版本

hbase(main):002:0> version

 

 

 ddl操作

 

1.創建一個member表

hbase(main):013:0> create 'table1','tab1_id','tab1_add','tab1_info'

 

 

2.查看所有的表

hbase(main):006:0> list

 

 

3.查看表結構

hbase(main):007:0> describe 'member'

 

 

4.刪除一個列簇

 

5、查看表是否存在

6、判斷表是否為"enable"

7、刪除一個表

dml操作

 1、創建member表

刪除一個列簇（一般不超過兩個列簇）

2、往member表插入數據

3、掃描查看數據

4、獲取數據

獲取一個rowkey的所有數據

獲取一個rowkey，一個列簇的所有數據

 

獲取一個rowkey，一個列簇中一個列的所有數據

 5、更新數據

6、刪除列簇中其中一列

7、統計表中總的行數

8、清空表中數據

 

2.java API

最常規且最高效的訪問方式

 

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.KeyValue;
import org.apache.hadoop.hbase.MasterNotRunningException;
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.util.Bytes;

public class HbaseTest {
    public static Configuration conf;
    static{
        conf = HBaseConfiguration.create();//第一步
        conf.set("hbase.zookeeper.quorum", "header-2,core-1,core-2");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        conf.set("hbase.master", "header-1:60000");
    }

    public static void main(String[] args) throws IOException{
        //createTable("member");
        //insertDataByPut("member");
        //QueryByGet("member");
        QueryByScan("member");
        //DeleteData("member");
    }
 


    /**
     * 創建表  通過HBaseAdmin對象操作
     * 
     * @param tablename
     * @throws IOException 
     * @throws ZooKeeperConnectionException 
     * @throws MasterNotRunningException 
     * 
     */
    public static void createTable(String tableName) throws MasterNotRunningException, ZooKeeperConnectionException, IOException {
        //創建HBaseAdmin對象
        HBaseAdmin hBaseAdmin = new HBaseAdmin(conf);
        //判斷表是否存在，若存在就刪除
        if(hBaseAdmin.tableExists(tableName)){
            hBaseAdmin.disableTable(tableName);
            hBaseAdmin.deleteTable(tableName);
        }
        HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);
        //添加Family
        tableDescriptor.addFamily(new HColumnDescriptor("info"));
        tableDescriptor.addFamily(new HColumnDescriptor("address"));
        //創建表
        hBaseAdmin.createTable(tableDescriptor);
        //釋放資源
        hBaseAdmin.close();
    }
    
    /**
     * 
     * @param tableName
     * @throws IOException 
     */
    @SuppressWarnings("deprecation")
    public static void insertDataByPut(String tableName) throws IOException {
        //第二步   獲取句柄，傳入靜態配置和表名稱
        HTable table = new HTable(conf, tableName);
        
        //添加rowkey，添加數據，  通過getBytes方法將string類型都轉化為字節流
        Put put1 = new Put(getBytes("djt"));
        put1.add(getBytes("address"), getBytes("country"), getBytes("china"));
        put1.add(getBytes("address"), getBytes("province"), getBytes("beijing"));
        put1.add(getBytes("address"), getBytes("city"), getBytes("beijing"));
        
        put1.add(getBytes("info"), getBytes("age"), getBytes("28"));
        put1.add(getBytes("info"), getBytes("birthdy"), getBytes("1998-12-23"));
        put1.add(getBytes("info"), getBytes("company"), getBytes("dajiang"));
        
        //第三步
        table.put(put1);
        
        //釋放資源
        table.close();
    }
    
    /**
     * 查詢一條記錄
     * @param tableName
     * @throws IOException 
     */
    public static void QueryByGet(String tableName) throws IOException {
        //第二步
        HTable table = new HTable(conf, tableName);
        //根據rowkey查詢
        Get get = new Get(getBytes("djt"));
        Result r = table.get(get);
        System.out.println("獲得到rowkey:" + new String(r.getRow()));
        for(KeyValue keyvalue : r.raw()){
            System.out.println("列簇：" + new String(keyvalue.getFamily())
            + "====列:" + new String(keyvalue.getQualifier())
            + "====值:" + new String(keyvalue.getValue()));
        }
        table.close();
    }
    


    /**
     * 掃描
     * @param tableName
     * @throws IOException 
     */
    public static void QueryByScan(String tableName) throws IOException {
        // 第二步
        HTable table = new HTable(conf, tableName);
        Scan scan = new Scan();
        //指定需要掃描的列簇，列.如果不指定就是全表掃描
        scan.addColumn(getBytes("info"), getBytes("company"));
        ResultScanner scanner = table.getScanner(scan);
        for(Result r : scanner){
            System.out.println("獲得到rowkey:" + new String(r.getRow()));
            for(KeyValue kv : r.raw()){
                System.out.println("列簇:" + new String(kv.getFamily())
                + "====列:" + new String(kv.getQualifier())
                + "====值 :" + new String(kv.getValue()));
            }
        }
        //釋放資源
        scanner.close();
        table.close();
    }
    
     

    /**
     * 刪除一條數據
     * @param tableName
     * @throws IOException 
     */
    public static void DeleteData(String tableName) throws IOException {
        // 第二步
        HTable table = new HTable(conf, tableName);
        
        Delete delete = new Delete(getBytes("djt"));
        delete.addColumn(getBytes("info"), getBytes("age"));
        
        table.delete(delete);
        //釋放資源
        table.close();
    }

    /**
     * 轉換byte數組(string類型都轉化為字節流)
     */
    public static byte[] getBytes(String str){
        if(str == null)
            str = "";
            return Bytes.toBytes(str);    
    }
}

 

 

3.MapReduce

直接使用MapReduce作業處理HBase數據

 

import java.io.IOException;
import java.util.StringTokenizer;

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.HBaseAdmin;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

/**
 * 將hdfs里面的數據導入hbase
 * @author Administrator
 *
 */
public class MapReduceWriteHbaseDriver {
    
    public static class WordCountMapperHbase extends Mapper<Object, Text, 
           ImmutableBytesWritable, IntWritable>{
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();
        
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException{
            StringTokenizer itr = new StringTokenizer(value.toString());
            while(itr.hasMoreTokens()){
                word.set(itr.nextToken());
                //輸出到hbase的key類型為ImmutableBytesWritable
                context.write(new ImmutableBytesWritable(Bytes.toBytes(word.toString())), one);
            }
        }
    }
    
    public static class WordCountReducerHbase extends TableReducer<ImmutableBytesWritable,
              IntWritable, ImmutableBytesWritable>{
        private IntWritable result = new IntWritable();
        public void reduce(ImmutableBytesWritable key, Iterable<IntWritable> values, Context context) 
                throws IOException, InterruptedException{
            int sum = 0;
            for(IntWritable val : values){
                sum += val.get();    
            }
            //put實例化 key代表主鍵，每個單詞存一行
            Put put = new Put(key.get());
            //三個參數分別為：列簇content  列count  列值為詞頻
            put.add(Bytes.toBytes("content"), Bytes.toBytes("count"), Bytes.toBytes(String.valueOf(sum)));
            context.write(key, put);
        }
    }
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException{
        String tableName = "wordcount";//hbase數據庫表名   也可以通過命令行傳入表名args
        Configuration conf = HBaseConfiguration.create();//實例化Configuration
        conf.set("hbase.zookeeper.quorum", "header-2,core-1,core-2");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        conf.set("hbase.master", "header-1");
        
        //如果表已經存在就先刪除
        HBaseAdmin admin = new HBaseAdmin(conf);
        if(admin.tableExists(tableName)){
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
        }
        
        HTableDescriptor htd = new HTableDescriptor(tableName);//指定表名
        HColumnDescriptor hcd = new HColumnDescriptor("content");//指定列簇名
        htd.addFamily(hcd);//創建列簇
        admin.createTable(htd);//創建表
        
        Job job = new Job(conf, "import from hdfs to hbase");
        job.setJarByClass(MapReduceWriteHbaseDriver.class);
        
        job.setMapperClass(WordCountMapperHbase.class);
        
        //設置插入hbase時的相關操作
        TableMapReduceUtil.initTableReducerJob(tableName, WordCountReducerHbase.class, job, null, null, null, null, false);
        
        //map輸出
        job.setMapOutputKeyClass(ImmutableBytesWritable.class);
        job.setMapOutputValueClass(IntWritable.class);
        
        //reduce輸出
        job.setOutputKeyClass(ImmutableBytesWritable.class);
        job.setOutputValueClass(Put.class);
        
        //讀取數據
        FileInputFormat.addInputPaths(job, "hdfs://header-1:9000/user/test.txt");
        System.out.println(job.waitForCompletion(true) ? 0 : 1);
    }
}

 

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 讀取hbse數據存入HDFS
 * @author Administrator
 *
 */
public class MapReduceReaderHbaseDriver {
    public static class WordCountHBaseMapper extends TableMapper<Text, Text>{
        protected void map(ImmutableBytesWritable key, Result values,Context context) throws IOException, InterruptedException{
            StringBuffer sb = new StringBuffer("");
            //獲取列簇content下面的值
            for(java.util.Map.Entry<byte[], byte[]> value : values.getFamilyMap("content".getBytes()).entrySet()){
                String str = new String(value.getValue());
                if(str != null){
                    sb.append(str);
                }
                context.write(new Text(key.get()), new Text(new String(sb)));
            }    
        }
    }

    public static class WordCountHBaseReducer extends Reducer<Text, Text, Text, Text>{
        private Text result = new Text();
        public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException{
            for(Text val : values){
                result.set(val);
                context.write(key, result);
            }
        }
    }
    
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        String tableName = "wordcount";//表名稱
        Configuration conf = HBaseConfiguration.create();//實例化Configuration
        conf.set("hbase.zookeeper.quorum", "header-2,core-1,core-2");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        conf.set("hbase.master", "header-1:60000");
        
        Job job = new Job(conf, "import from hbase to hdfs");
        job.setJarByClass(MapReduceReaderHbaseDriver.class);
        
        job.setReducerClass(WordCountHBaseReducer.class);
        //配置讀取hbase的相關操作
        TableMapReduceUtil.initTableMapperJob(tableName, new Scan(), WordCountHBaseMapper.class, Text.class, Text.class, job, false);
        
        //輸出路徑
        FileOutputFormat.setOutputPath(job, new Path("hdfs://header-1:9000/user/out"));
        System.out.println(job.waitForCompletion(true) ? 0 : 1);
    }    
}