package thathbase;

import java.io.IOException;
import java.util.Random;

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

public class HelloHbase {
    static Configuration conf;
    static HBaseAdmin admin;
    static Connection conn;  
    static HTableDescriptor tableDescriptor;
    static HTable table;
    static Put putRow1;
    static Put putRow2;
    static Random rand =new Random(25);

    public static void main(String[] args) throws Exception {
        // TODO Auto-generated method stub
        init();
        createTable();
        insertTable();

    }
    
    private static void init() throws Exception {
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        conf.set("hbase.zookeeper.quorum", "master,slave03,slave04");
        conf.set("hbase.master", "master:60000");                 
        conn = ConnectionFactory.createConnection(conf);
        admin = (HBaseAdmin) conn.getAdmin();
    }
    
    @SuppressWarnings("deprecation")
    private static void createTable() throws Exception{
        admin.disableTable("scores");
        admin.deleteTable("scores");
        if (!admin.tableExists("scores")){
            tableDescriptor = new HTableDescriptor("scores".getBytes());
            tableDescriptor.addFamily(new HColumnDescriptor("fam1"));
            admin.createTable(tableDescriptor);    
        }
        else{
            System.out.println("Table already exists!");
        }
        table = new HTable(conf, "scores");                
    }
    
    @SuppressWarnings("deprecation")
    private static void insertTable() throws IOException{
        putRow1 = new Put("row1".getBytes());
        putRow2 = new Put("row2".getBytes());
        
        for (int i =0; i<1000; i++){
            String row = String.valueOf(i);
            if (rand1() == 1){
                putRow1 = new Put(row.getBytes());
                putRow1.add("fam1".getBytes(),"col1".getBytes(),String.valueOf(rand2()).getBytes());
                table.put(putRow1);
                System.out.println(i + ":Insert into col1.");
            }
            else{
                putRow2 = new Put(row.getBytes());
                putRow2.add("fam1".getBytes(),"col2".getBytes(),String.valueOf(rand2()).getBytes());
                table.put(putRow2);
                System.out.println(i + ":Insert into col2.");
            }
        }
    }
    
    private static int rand1(){
        int r = rand.nextInt(100);
        if (r < 50)
            return 1;
        else
            return 2;
    }
    
    private static int rand2(){
        int r = rand.nextInt(10000);
        return r;
    }
}