Index: /sample/HBaseRecordPro.java
===================================================================
--- /sample/HBaseRecordPro.java	(revision 16)
+++ /sample/HBaseRecordPro.java	(revision 17)
@@ -8,50 +8,57 @@
 /**
  * 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,9 +68,9 @@
 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,86 +87,78 @@
 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,5 +179,17 @@
 		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,48 +197,35 @@
 	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,4 +234,7 @@
 		JobClient.runJob(conf);
 		
+		// delete tmp file
+		FileSystem.get(conf).delete(text_path);
+		setup.deleteFile(conf_tmp);
 	}
 }