package com.beifeng.senior.hadoop.hbase;

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.client.Delete;

import org.apache.hadoop.hbase.client.Get;

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;

import org.apache.hadoop.io.IOUtils;

/**

 * CRUD operations

 *

 * @author root

 *

 */

public class HBaseOperation {

    public static HTable getHTableByTableName(String tableName) throws Exception {

        // get instance of default configuration

        Configuration configuration = HBaseConfiguration.create();

        // get table instance

        HTable table = new HTable(configuration, tableName);

        return table;

    }

    public void getData() throws Exception {

        String tableName = "user"; // default.user

        HTable table = getHTableByTableName(tableName);

        //create get with rowkey

        Get get = new Get(Bytes.toBytes("10002"));

        //add column

        get.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"));

        get.addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"));

        //get data

        Result result = table.get(get);

        //key: rowkey + cf + c + version

        //value: value

        for(Cell cell : result.rawCells()){

            System.out.println(

                    Bytes.toString(CellUtil.cloneFamily(cell)) + ":" //

                    + Bytes.toString(CellUtil.cloneQualifier(cell)) + "->" //

                    + Bytes.toString(CellUtil.cloneValue(cell)));

        }

        //table close

        table.close();

    }

    /**

     * 建议：

     *         tablename & column family ->常量, HbaseTableContent

     *

     *         Map<String, Object>

     *

     * @throws Exception

     */

    public void putData() throws Exception {

        String tableName = "user"; // default.user

        HTable table = getHTableByTableName(tableName);

        Put put = new Put(Bytes.toBytes("10004"));

        //add a column with value

        put.add(Bytes.toBytes("info"),

                 Bytes.toBytes("name"),

                 Bytes.toBytes("zhaoliu"));

        put.add(Bytes.toBytes("info"),

                 Bytes.toBytes("age"),

                 Bytes.toBytes("25"));

        put.add(Bytes.toBytes("info"),

                 Bytes.toBytes("address"),

                 Bytes.toBytes("shanghai"));

        table.put(put);

        table.close();

    }

    public void deleteData() throws Exception {

        String tableName = "user"; // default.user

        HTable table = getHTableByTableName(tableName);

        Delete delete = new Delete(Bytes.toBytes("10004"));

        /*删除单个数据

        delete.deleteColumn(Bytes.toBytes("info"),

                                 Bytes.toBytes("address"));

        */

        /*删除整个列簇*/

        delete.deleteFamily(Bytes.toBytes("info"));

        table.delete(delete);

        table.close();

    }

    public static void main(String[] args) throws Exception {

        String tableName = "user"; // default.user

        HTable table = null;

        ResultScanner resultScanner = null;

        try {

            table = getHTableByTableName(tableName);

            /*全表扫描

            Scan scan = new Scan();

            */

            //指定Rowkey范围

            Scan scan = new Scan();

            scan.setStartRow(Bytes.toBytes("10002"));

            scan.setStopRow(Bytes.toBytes("10004"));

            resultScanner = table.getScanner(scan);

            for(Result result : resultScanner) {

                System.out.println(Bytes.toString(result.getRow()));

                //System.out.println(result);

                for(Cell cell : result.rawCells()){

                    System.out.println(

                            Bytes.toString(CellUtil.cloneFamily(cell)) + ":" //

                            + Bytes.toString(CellUtil.cloneQualifier(cell)) + "->" //

                            + Bytes.toString(CellUtil.cloneValue(cell)));

                }

                System.out.println("--------------------");

            }

        } catch (Exception e) {

            e.printStackTrace();

        }finally{

            IOUtils.closeStream(resultScanner);

            IOUtils.closeStream(table);

        }

    }

}

hbase(main):006:0> flush 'table name'        //刷数据

hbase(main):007:0> compact 'table name'    //合并数据