package itri;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;

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;


// ITRI serial program number 1 

// 1. create table then read file to insert into hbase
//    tablename: itri 
//    family : Detail, Products, Turnover
// 2. edit a text file
//    format of "/tmp/itri/store.txt" is as
//    T01;GunLong;  01;20;40;30;50
//    T02;Esing;  02;50
//    T03;SunDon; 03;40;30
//    T04;StarBucks;  04;50;50;20
// 
public class Itri1LoadFile {

  void loadFile2HBase(String file_in, String table_name) throws IOException {

    BufferedReader fi = new BufferedReader(
        new FileReader(new File(file_in)));

    String line;
    while ((line = fi.readLine()) != null) {

      String[] str = line.split(";");
      int length = str.length;
      PutData.putData(table_name, str[0].trim(), "Detail", "Name", str[1]
          .trim());
      PutData.putData(table_name, str[0].trim(), "Detail", "Locate",
          str[2].trim());
      for (int i = 3; i < length; i++) {
        PutData.putData(table_name, str[0], "Products", "P" + (i - 2),
            str[i]);
      }
      System.out.println();
    }
    fi.close();

  }

  public void createHBaseTable(String tablename, String[] family)
      throws IOException {

    HTableDescriptor htd = new HTableDescriptor(tablename);

    for (String fa : family) {
      htd.addFamily(new HColumnDescriptor(fa));
    }

    HBaseConfiguration config = new HBaseConfiguration();

    HBaseAdmin admin = new HBaseAdmin(config);

    if (admin.tableExists(tablename)) {
      System.out.println("Table: " + tablename + "Existed.");
    } else {
      System.out.println("create new table: " + tablename);

      admin.createTable(htd);
    }
  }

  static public void main(String args[]) throws IOException {
    String tablename = "itri";
    String[] family = { "Detail", "Products", "Turnover" };
    String loadfile = "/tmp/itri/store.txt";
    // run
    Itri1LoadFile lf = new Itri1LoadFile();
    lf.createHBaseTable(tablename, family);
    lf.loadFile2HBase(loadfile, tablename);
  }
}