Changeset 17

Timestamp:

Jul 1, 2008, 11:22:41 AM (16 years ago)

Author:

waue

Message:

more comments; much readable

File:

• sample/HBaseRecordPro.java (modified) (6 diffs)

sample/HBaseRecordPro.java

@@ -8 +8 @@
 /**
  * Purpose : 
- *  Parse your record and then store in HBase.
+ *  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 Hadoop file system and Hbase are running correctly.
- *  1. put test.txt in t1 directory which content is 
- ---------------
- name:locate:years 
- waue:taiwan:1981
- shellon:taiwan:1981
- ---------------
- *  2. hadoop_root/$ bin/hadoop dfs -put t1 t1
- *  3. hbase_root/$ bin/hbase shell
- *  4. hql > create table t1_table("person");
- *  5. Come to Eclipse and run this code, and we will let database as that 
- t1_table -> person
- ----------------
- |  name | locate | years |
- | waue  | taiwan | 1981 |
- | shellon | taiwan | 1981 |
- ----------------
+ *  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                       | person:locate           | locate                  |
- +-------------------------+-------------------------+-------------------------+
- | 0                       | person:name             | name                    |
- +-------------------------+-------------------------+-------------------------+
- | 0                       | person:years            | years                   |
- +-------------------------+-------------------------+-------------------------+
- | 19                      | person:locate           | taiwan                  |
- +-------------------------+-------------------------+-------------------------+
- | 19                      | person:name             | waue                    |
- +-------------------------+-------------------------+-------------------------+
- | 19                      | person:years            | 1981                    |
- +-------------------------+-------------------------+-------------------------+
- | 36                      | person:locate           | taiwan                  |
- +-------------------------+-------------------------+-------------------------+
- | 36                      | person:name             | shellon                 |
- +-------------------------+-------------------------+-------------------------+
- | 36                      | person:years            | 1981                    |
- +-------------------------+-------------------------+-------------------------+
- 3 row(s) in set. (0.04 sec)
+
++-------------------------+-------------------------+-------------------------+
+| 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)
+
  */
 
@@ -61 +68 @@
 import java.io.BufferedWriter;
 import java.io.File;
-import java.io.FileOutputStream;
 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;
@@ -80 +87 @@
 import org.apache.hadoop.mapred.lib.IdentityReducer;
 
-
-
-class ReduceClass extends TableReduce<LongWritable, Text> {
+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
-  public void reduce(LongWritable key, Iterator<Text> values,
-      OutputCollector<Text, MapWritable> output, Reporter reporter)
+  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 {
-    String sp = ":";
-    String bf = "person:";
-    // this map holds the columns per row
-    MapWritable map = new MapWritable();
-    // values.next().getByte() can get value and transfer to byte form,
-    String stro = new String(values.next().getBytes());
-    String str[] = stro.split(sp);
-    
-    BufferedReader fconf = new BufferedReader(new FileReader(new File("/tmp/fi_conf.tmp")));
-    // int length = cf.length;
-    
-    
-    
-    // test for debug
-    FileOutputStream out = new FileOutputStream(new File(
-        "/home/waue/mr-result.txt"));
-    String first_line = fconf.readLine();
-    
-    // test for debug
-    String[] cf = first_line.split(sp);
-    int length = cf.length;
-    for(int i=0 ; i<length; i ++){
-      out.write((bf + cf[i]+"\n").getBytes());
-      
-    }
-    out.close();
-    // 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];
-    map.clear();
-
-    for (int i = 0; i < length; i++) {
-      col_n[i] = new Text(bf + 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 class HBaseRecordPro {
-
-  /* Denify parameter */
-
-  // file path in hadoop file system (not phisical file system)
-  final String file_path = "/home/waue/test.txt";
-
-  // setup MapTask and Reduce Task
-
-  final String bf = "person:";
-
-  final String table_name = "testend";
-
-  final String sp = ":";
-
-  String[] cf ;
-  
-  String test;
-
-  public HBaseRecordPro() {
-
-
-  }
-  public HBaseRecordPro(String[] st) {
-    cf = st;
-  }
-
-  static 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("/tmp/fi_conf.tmp")));
+    BufferedWriter ff = new BufferedWriter(new FileWriter(new File(conf_tmp)));
     BufferedWriter fw = new BufferedWriter(new FileWriter(new File(ou)));
     String first_line, data;
@@ -180 +179 @@
     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.
@@ -186 +197 @@
   public static void main(String[] args) throws IOException {
 
-    String bf = "person:";
-    String file_path = "/home/waue/test.txt";
+    HBaseRecordPro setup = new HBaseRecordPro();
+    String[] col_family = {column_family};
+    Path text_path = new Path(text_tmp);
     
-    final String file_tmp = "/tmp/HBaseRecord.text.tmp";
-    final int mapTasks = 1;
-    final int reduceTasks = 1;
-    String[] column_family = { bf };
-    
-    HBaseRecordPro setup = new HBaseRecordPro();
-    
-    String first_line = parseFirstLine(file_path, file_tmp);
-    System.out.println(first_line);
-//    HBaseRecordPro.cf = first_line.split(sp);
-
-    
-    // test
-    /*
-    for (int i = 0; i < 3; i++) {
-      System.out.println("column[" + i + "]=" + bf + cf[i]);
-    }*/
-
-    BuildHTable build_table = new BuildHTable(setup.table_name,
-        column_family);
-    if (!build_table.checkTableExist(setup.table_name)) {
+    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 \"" + setup.table_name
+      System.out.println("Table \"" + table_name
           + "\" has already existed !");
     }
     JobConf conf = new JobConf(HBaseRecordPro.class);
     FileSystem fileconf = FileSystem.get(conf);
-    fileconf.copyFromLocalFile(true, new Path(file_tmp), new Path(file_tmp));
+    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(setup.table_name, ReduceClass.class, conf);
-    
+    TableReduce.initJob(table_name, ReduceClass.class, conf);
+
     // below are map-reduce profile
     conf.setNumMapTasks(mapTasks);
     conf.setNumReduceTasks(reduceTasks);
-    conf.setInputPath(new Path(file_tmp));
+    conf.setInputPath(text_path);
     conf.setMapperClass(IdentityMapper.class);
     conf.setCombinerClass(IdentityReducer.class);
@@ -236 +234 @@
     JobClient.runJob(conf);
     
+    // delete tmp file
+    FileSystem.get(conf).delete(text_path);
+    setup.deleteFile(conf_tmp);
   }
 }