source: sample/hadoop-0.16/tw/org/nchc/code/HBaseRecordPro.java @ 27

/**
 * Program: HBaseRecordPro.java
 * Editor: Waue Chen 
 * From :  NCHC. Taiwn
 * Last Update Date: 07/02/2008
 */

/**
 * Purpose : 
 *  First, program would parse your record and create Hbase.\
 *  Then it sets the first line as column qualify \
 *  Finally it stores in HBase automatically.
 * 
 * HowToUse : 
 *  Make sure two thing :
 *  1. source_file must be regular as follow:
 *    first line: qualify1:qualify2:...:qualifyN
 *    other line: records1:records2:...:recordsN
 *     (the number N of qualify must be the same as records ) 
-----------------
name:locate:years
waue:taiwan:1981
rock:taiwan:1981
aso:taiwan:1981
jazz:taiwan:1982
-----------------
 *  2. source_file path must be correct.

 * Check Result:
 *  Go to hbase console, type : 
 *    hql > select * from t1_table; 
 08/06/06 12:20:48 INFO hbase.HTable: Creating scanner over t1_table starting at key 

+-------------------------+-------------------------+-------------------------+
| Row                     | Column                  | Cell                    |
+-------------------------+-------------------------+-------------------------+
| 0                       | member:locate           | taiwan                  |
+-------------------------+-------------------------+-------------------------+
| 0                       | member:name             | waue                    |
+-------------------------+-------------------------+-------------------------+
| 0                       | member:years            | 1981                    |
+-------------------------+-------------------------+-------------------------+
| 17                      | member:locate           | taiwan                  |
+-------------------------+-------------------------+-------------------------+
| 17                      | member:name             | rock                    |
+-------------------------+-------------------------+-------------------------+
| 17                      | member:years            | 1981                    |
+-------------------------+-------------------------+-------------------------+
| 34                      | member:locate           | taiwan                  |
+-------------------------+-------------------------+-------------------------+
| 34                      | member:name             | aso                     |
+-------------------------+-------------------------+-------------------------+
| 34                      | member:years            | 1981                    |
+-------------------------+-------------------------+-------------------------+
| 50                      | member:locate           | taiwan                  |
+-------------------------+-------------------------+-------------------------+
| 50                      | member:name             | jazz                    |
+-------------------------+-------------------------+-------------------------+
| 50                      | member:years            | 1982                    |
+-------------------------+-------------------------+-------------------------+
4 row(s) in set. (0.31 sec)

 */

package tw.org.nchc.code;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapred.TableReduce;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.lib.IdentityMapper;
import org.apache.hadoop.mapred.lib.IdentityReducer;

public class HBaseRecordPro {
  /* Major parameter */
  // it indicates local path, not hadoop file system path
  final static String source_file = "/home/waue/test.txt";

  /* Minor parameter */
  // column family name
  final static String column_family = "member:";

  // table name
  final static String table_name = "HBaseRecord";

  // separate char
  final static String sp = ":";
  
  // conf tmp with column qualify
  final static String conf_tmp = "/tmp/HBaseRecordPro.firstLine.tmp";
  
  // data source tmp
  final static String text_tmp = "/tmp/HBaseRecord.text.tmp";

  // on this sample, map is nonuse, we use reduce to handle
  private static class ReduceClass extends TableReduce<LongWritable, Text> {
    public void reduce(LongWritable key, Iterator<Text> values,
        OutputCollector<Text, MapWritable> output, Reporter reporter)
        throws IOException {

      // read the configure file
      BufferedReader fconf = new BufferedReader(new FileReader(new File(
          conf_tmp)));
      String first_line = fconf.readLine();
      fconf.close();
      // extract cf data
      String[] cf = first_line.split(sp);
      int length = cf.length;
       
      // values.next().getByte() can get value and transfer to byte form,
      String stro = new String(values.next().getBytes());
      String str[] = stro.split(sp);

      // Column id is created dymanically,
      Text[] col_n = new Text[length];
      byte[][] b_l = new byte[length][];
      // contents must be ImmutableBytesWritable
      ImmutableBytesWritable[] w_l = new ImmutableBytesWritable[length];

      // This map connect to hbase table and holds the columns per row
      MapWritable map = new MapWritable();
      map.clear();

      // prepare to write data into map
      for (int i = 0; i < length; i++) {
        col_n[i] = new Text(column_family + cf[i]);
        b_l[i] = str[i].getBytes();
        w_l[i] = new ImmutableBytesWritable(b_l[i]);
        // populate the current row
        map.put(col_n[i], w_l[i]);
      }
      // add the row with the key as the row id
      output.collect(new Text(key.toString()), map);
    }
  }

  public HBaseRecordPro() {
  }
  
  // This function can split the source text into two file, \
  //  conf_tmp file recorded first line is used to set column qualify \
  //  text_tmp , ou, recorded data is used to store into table.
  public String parseFirstLine(String in, String ou)
      throws IOException {

    BufferedReader fi = new BufferedReader(new FileReader(new File(in)));
    BufferedWriter ff = new BufferedWriter(new FileWriter(new File(conf_tmp)));
    BufferedWriter fw = new BufferedWriter(new FileWriter(new File(ou)));
    String first_line, data;
    first_line = fi.readLine();
    ff.write(first_line);
    ff.flush();
    do {
      data = fi.readLine();
      if (data == null) {
        break;
      } else {
        fw.write(data + "\n");
        fw.flush();
      }
    } while (true);
    fw.close();
    ff.close();
    return first_line;
  }
  // tmp file delete
  boolean deleteFile(String str)throws IOException{
    File df = new File(str);
    
    if(df.exists()){
      if(!df.delete()){
        System.err.print("delete file error !");
      }
    }else{
      System.out.println("file not exit!");
    }
    return true;
  }
  /**
   * Runs the demo.
   */
  public static void main(String[] args) throws IOException {

    HBaseRecordPro setup = new HBaseRecordPro();
    String[] col_family = {column_family};
    Path text_path = new Path(text_tmp);
    
    setup.parseFirstLine(source_file, text_tmp);
//    System.out.println(first_line);

    BuildHTable build_table = new BuildHTable(table_name,
        col_family);
    if (!build_table.checkTableExist(table_name)) {
      if (!build_table.createTable()) {
        System.out.println("create table error !");
      }
    } else {
      System.out.println("Table \"" + table_name
          + "\" has already existed !");
    }
    JobConf conf = new JobConf(HBaseRecordPro.class);
    FileSystem fileconf = FileSystem.get(conf);
    fileconf.copyFromLocalFile(true, text_path, text_path);
    // Job name; you can modify to any you like
    conf.setJobName("PersonDataBase");
    final int mapTasks = 1;
    final int reduceTasks = 1;
    // Hbase table name must be correct , in our profile is t1_table
    TableReduce.initJob(table_name, ReduceClass.class, conf);

    // below are map-reduce profile
    conf.setNumMapTasks(mapTasks);
    conf.setNumReduceTasks(reduceTasks);
    
    conf.setInputPath(text_path);

    
    conf.setMapperClass(IdentityMapper.class);
    conf.setCombinerClass(IdentityReducer.class);
    conf.setReducerClass(ReduceClass.class);

    JobClient.runJob(conf);
    
    // delete tmp file
    // 0.16
    FileSystem.get(conf).delete(text_path);
    
    setup.deleteFile(conf_tmp);
  }
}