Index: mple/BuildHTable.java =================================================================== --- /sample/BuildHTable.java (revision 22) +++ (revision ) @@ -1,70 +1,0 @@ -/** - * Program: BuildHTable.java - * Editor: Waue Chen - * From : NCHC. Taiwn - * Last Update Date: 06/10/2008 - */ - -package tw.org.nchc.code; -import java.io.IOException; -import org.apache.hadoop.hbase.HBaseAdmin; -import org.apache.hadoop.hbase.HBaseConfiguration; -import org.apache.hadoop.hbase.HColumnDescriptor; -import org.apache.hadoop.hbase.HTableDescriptor; -import org.apache.hadoop.io.Text; - -public class BuildHTable { - private String table_name ; - private String[] Column_Family ; - HBaseConfiguration conf = new HBaseConfiguration(); - HBaseAdmin admin = new HBaseAdmin(conf); - public BuildHTable(String table,String[] CF) throws IOException{ - table_name = table; - Column_Family = CF; - } - - - public boolean checkTableExist(String tname)throws IOException{ - if(! admin.tableExists(new Text(tname))){ - return false; - } - return true; - } - // create Hbase table , success = 1 ; failse = 0; Table_exist = 2; - public boolean createTable() throws IOException{ - // check whether Table name exite or not - if(! checkTableExist(table_name)){ - System.out.println("HTable : " + table_name + " creating ... please wait"); - HTableDescriptor tableDesc = new HTableDescriptor(table_name); - for(int i =0; i" +Column_Family[i]); - } - //add column family - tableDesc.addFamily(new HColumnDescriptor(Column_Family[i])); - } - admin.createTable(tableDesc); - } else { - return false; - } - return true; - } - public static void main(String[] args) throws IOException { - - // setup Table name - String Table_Name = "test_create_table2"; - // setup Column Family - String[] Column_Family = {"http:","url:","referrer:"}; - - BuildHTable bt = new BuildHTable( Table_Name , Column_Family); - boolean ret = bt.createTable(); - if(ret == true){ - System.out.println("Create Table \"" +Table_Name +" \" Compelte !!!"); - }else { - System.out.println("Table Name \"" +Table_Name +" \" exit!!!"); - } - } -} Index: mple/Convert.java =================================================================== --- /sample/Convert.java (revision 22) +++ (revision ) @@ -1,33 +1,0 @@ -package tw.org.nchc.code; -import java.io.IOException; - -import org.apache.hadoop.fs.FileStatus; -import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.fs.Path; -import org.apache.hadoop.mapred.FileInputFormat; -import org.apache.hadoop.mapred.FileOutputFormat; -import org.apache.hadoop.mapred.JobConf; - -public class Convert -{ - static public Path[] listPaths(FileSystem fsm,Path path) throws IOException - { - FileStatus[] fss = fsm.listStatus(path); - int length = fss.length; - Path[] pi = new Path[length]; - for (int i=0 ; i< length; i++) - { - pi[i] = fss[i].getPath(); - } - return pi; - } - static public void setInputPath(JobConf conf, Path path){ - FileInputFormat.setInputPaths(conf, path); - } - static public void addInputPath(JobConf conf, Path path){ - FileInputFormat.addInputPath(conf, path); - } - static public void setOutputPath(JobConf conf,Path path){ - FileOutputFormat.setOutputPath(conf, path); - } -} Index: mple/HBaseClient.java =================================================================== --- /sample/HBaseClient.java (revision 22) +++ (revision ) @@ -1,152 +1,0 @@ -/** - * Program: HBaseClient.java - * Editor: Waue Chen - * From : NCHC. Taiwn - * Last Update Date: 06/10/2008 - */ - -package tw.org.nchc.code; - -import java.io.IOException; -import java.util.SortedMap; -import java.util.TreeMap; - -import org.apache.hadoop.hbase.HBaseConfiguration; -import org.apache.hadoop.hbase.HScannerInterface; -import org.apache.hadoop.hbase.HStoreKey; -import org.apache.hadoop.hbase.HTable; -import org.apache.hadoop.io.Text; - -/** - * Demo that illustrates the HBase client API. - * The demo program will insert values to " Column Family: Column qualifier" and then print these. - * pre-do : create a hbase table "test_table" with (CF:CI) which (" column family ", " column qualifier ") - * 1. $ bin/hbase shell - * 2. > create table test_table("CF"); - * ok ! we can test it - * 3. > insert into test_table("CF:CI") values=("Hellow World") where row = "1"; - * 4. > select * from test_table; - -08/06/03 16:16:36 INFO hbase.HTable: Creating scanner over test_table starting at key -+---------+-----------+-----------+ -| Row | Column | Cell | -+---------+-----------+-----------+ -| 1 | CF:CI | Hellow World | -+---------+-----------+-----------+ -1 row(s) in set. (0.24 sec) - - * on the structure , "Row" means Row ID which is a key to describe a column; - * Column means the database structure in test_table, - * Column Family , "CF", should be defined while creating table. - * Column qualifier , "CI" , can be added dynamically. - * Cell is the value of CF:CI - * - * that's the structure; then the demo program will show you in console as below : - * -Illustration of adding data... -Writing row = 0, col 'CF:CI' = Hellow0 -Writing row = 1, col 'CF:CI' = Hellow1 -Writing row = 2, col 'CF:CI' = Hellow2 -Writing row = 3, col 'CF:CI' = Hellow3 -Writing row = 4, col 'CF:CI' = Hellow4 -Writing row = 5, col 'CF:CI' = Hellow5 -Writing row = 6, col 'CF:CI' = Hellow6 -Writing row = 7, col 'CF:CI' = Hellow7 -Writing row = 8, col 'CF:CI' = Hellow8 -Writing row = 9, col 'CF:CI' = Hellow9 - -Illustration of querying... -row = 1, 'CF : CI ' = Hellow1 - -Illustration of scanning... -08/06/03 16:47:51 INFO hbase.HTable: Creating scanner over test_table starting at key -row = 0//9223372036854775807, col 'CF:CI' = Hellow0 -row = 1//9223372036854775807, col 'CF:CI' = Hellow1 -row = 2//9223372036854775807, col 'CF:CI' = Hellow2 -row = 3//9223372036854775807, col 'CF:CI' = Hellow3 -row = 4//9223372036854775807, col 'CF:CI' = Hellow4 -row = 5//9223372036854775807, col 'CF:CI' = Hellow5 -row = 6//9223372036854775807, col 'CF:CI' = Hellow6 -row = 7//9223372036854775807, col 'CF:CI' = Hellow7 -row = 8//9223372036854775807, col 'CF:CI' = Hellow8 -row = 9//9223372036854775807, col 'CF:CI' = Hellow9 - - - * - */ -public class HBaseClient { - - public static void main(String[] args) throws IOException { - - // Open the "test_table" table. If it hasn't been in Hbase, you should create. - HBaseConfiguration conf = new HBaseConfiguration(); - HTable table = new HTable(conf, new Text("test_table")); - - System.out.println("Illustration of adding data..."); - - // create column formed (Column Family:Column qualifier) - Text column = new Text("CF:CI"); - - // create row_id - Text row_id = new Text(); - - // demo 1 : Insert ten demo values - for (int i = 0; i < 10; i++) { - - // give row_id value - row_id.set(new Integer(i).toString()); - - // let "indicate_id" indicate the column which row = row_id - long indicate_id= table.startUpdate(row_id); - - //val = value of CF:CI where row_id = i - Text val = new Text("Hellow" + i); - - // put "val" to "column" from "table" where "row_id" - // the same as : - // hql> INSERT INTO table( column ) VALUES=( val) WHERE ROW = row_id ; - table.put(indicate_id, column, val.getBytes()); - table.commit(indicate_id); - - System.out.println("Writing row = " + row_id + ", col '" + column - + "' = " + val); - } - - // demo 2 : print column value only row = 1 ; - System.out.println("\n Querying row = 1"); - - // Get a single value for the specified row and column - // byte[] = HTable.get(Text row, Text column) - - String s = Text.decode(table.get(new Text("1"),new Text("CF:CI"))); - // if change as - // String s = (table.get(new Text("1"),new Text("CF:CI"))).toString(); - // will get chaos code " [B@1f14ceb" - System.out.println("row = 1, 'CF : CI ' = " + s); - - // demo 3 : Print the all contents of this table - System.out.println("\nIllustration of scanning..."); - - // we only want one column, but you can specify multiple columns to - // fetch at once - Text[] cols = { column }; - - // Use HScannerInterface to crawl table - HScannerInterface scanner = table.obtainScanner(cols, new Text()); - - // column values are stored in a Map - SortedMap values = new TreeMap(); - HStoreKey currentKey = new HStoreKey(); - while (scanner.next(currentKey, values)) { - // decode the stored byte[] back into a String - String val = Text.decode(values.get(column)); - System.out.println("row = " + currentKey + ", col '" + column + "' = " - + val); - } - - // remember to close scanner when done - scanner.close(); - - } - -} Index: mple/HBaseRecord.java =================================================================== --- /sample/HBaseRecord.java (revision 22) +++ (revision ) @@ -1,174 +1,0 @@ -/** - * Program: HBaseRecord.java - * Editor: Waue Chen - * From : NCHC. Taiwn - * Last Update Date: 07/02/2008 - * Upgrade to 0.17 - */ - -/** - * Purpose : - * Parse your record and then store in HBase. - * - * 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 | - ---------------- - * 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) - */ - -package tw.org.nchc.code; - -import java.io.IOException; -import java.util.Iterator; - -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 HBaseRecord { - - /* Denify parameter */ - // one column family: person; three column qualifier: name,locate,years - static private String baseId1 = "person:name"; - - static private String baseId2 = "person:locate"; - - static private String baseId3 = "person:years"; - - // split character - static private String sp = ":"; - - // file path in hadoop file system (not phisical file system) - String file_path = "/user/waue/t1"; - - // Hbase table name - String table_name = "t1_table"; - - // setup MapTask and Reduce Task - int mapTasks = 1; - - int reduceTasks = 1; - - private static class ReduceClass extends TableReduce { - - // Column id is created dymanically, - private static final Text col_name = new Text(baseId1); - - private static final Text col_local = new Text(baseId2); - - private static final Text col_year = new Text(baseId3); - - // this map holds the columns per row - private MapWritable map = new MapWritable(); - - // on this sample, map is nonuse, we use reduce to handle - public void reduce(LongWritable key, Iterator values, - OutputCollector output, Reporter reporter) - throws IOException { - - // values.next().getByte() can get value and transfer to byte form, - // there is an other way that let decode() - // to substitude getByte() - String stro = new String(values.next().getBytes()); - String str[] = stro.split(sp); - byte b_local[] = str[0].getBytes(); - byte b_name[] = str[1].getBytes(); - byte b_year[] = str[2].getBytes(); - - // contents must be ImmutableBytesWritable - ImmutableBytesWritable w_local = new ImmutableBytesWritable(b_local); - ImmutableBytesWritable w_name = new ImmutableBytesWritable(b_name); - ImmutableBytesWritable w_year = new ImmutableBytesWritable(b_year); - - // populate the current row - map.clear(); - map.put(col_name, w_local); - map.put(col_local, w_name); - map.put(col_year, w_year); - - // add the row with the key as the row id - output.collect(new Text(key.toString()), map); - } - } - - private HBaseRecord() { - } - - /** - * Runs the demo. - */ - public static void main(String[] args) throws IOException { - // which path of input files in Hadoop file system - - HBaseRecord setup = new HBaseRecord(); - JobConf conf = new JobConf(HBaseRecord.class); - - // Job name; you can modify to any you like - conf.setJobName("NCHC_PersonDataBase"); - - // Hbase table name must be correct , in our profile is t1_table - TableReduce.initJob(setup.table_name, ReduceClass.class, conf); - - // below are map-reduce profile - conf.setNumMapTasks(setup.mapTasks); - conf.setNumReduceTasks(setup.reduceTasks); - - // 0.16 - // conf.setInputPath(new Path(setup.file_path)); - Convert.setInputPath(conf, new Path(setup.file_path)); - - conf.setMapperClass(IdentityMapper.class); - conf.setCombinerClass(IdentityReducer.class); - conf.setReducerClass(ReduceClass.class); - JobClient.runJob(conf); - } -} Index: mple/HBaseRecord2.java =================================================================== --- /sample/HBaseRecord2.java (revision 22) +++ (revision ) @@ -1,162 +1,0 @@ -/** - * Program: HBaseRecord.java - * Editor: Waue Chen - * From : NCHC. Taiwn - * Last Update Date: 07/01/2008 - * Upgrade to 0.17 - */ - -/** - * Purpose : - * Parse your record and then store in HBase. - * - * 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 - ---------------- - * 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) - */ - - - - -package tw.org.nchc.code; - -import java.io.IOException; -import java.util.Iterator; - -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 HBaseRecord2 { - - /* Denify parameter */ - static String[] bf = {"person:name","person:local","person:birthyear"}; - // file path in hadoop file system (not phisical file system) - String file_path = "/user/waue/t1/test.txt"; - // Hbase table name - String table_name = "testtable"; - - - // setup MapTask and Reduce Task - int mapTasks = 1; - int reduceTasks = 1; - - private static class ReduceClass extends TableReduce { - - - - // on this sample, map is nonuse, we use reduce to handle - public void reduce(LongWritable key, Iterator values, - OutputCollector output, Reporter reporter) - throws IOException { - // 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(":"); - - int length = bf.length; - - // 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[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); - } - } - - private HBaseRecord2() { - } - - /** - * Runs the demo. - */ - public static void main(String[] args) throws IOException { - - - HBaseRecord2 setup = new HBaseRecord2(); - String[] tmp = bf[0].split(":"); - String[] CF = {tmp[0]}; - BuildHTable build_table = new BuildHTable(setup.table_name, CF); - if (!build_table.checkTableExist(setup.table_name)) { - if (!build_table.createTable()) { - System.out.println("create table error !"); - } - } else { - System.out.println("Table \"" + setup.table_name - + "\" has already existed !"); - } - - JobConf conf = new JobConf(HBaseRecord2.class); - - //Job name; you can modify to any you like - conf.setJobName("PersonDataBase"); - - // Hbase table name must be correct , in our profile is t1_table - TableReduce.initJob(setup.table_name, ReduceClass.class, conf); - - // below are map-reduce profile - conf.setNumMapTasks(setup.mapTasks); - conf.setNumReduceTasks(setup.reduceTasks); - // 0.16 -// conf.setInputPath(new Path(setup.file_path)); - Convert.setInputPath(conf, new Path(setup.file_path)); - conf.setMapperClass(IdentityMapper.class); - conf.setCombinerClass(IdentityReducer.class); - conf.setReducerClass(ReduceClass.class); - JobClient.runJob(conf); - } -} Index: mple/HBaseRecordPro.java =================================================================== --- /sample/HBaseRecordPro.java (revision 22) +++ (revision ) @@ -1,247 +1,0 @@ -/** - * Program: HBaseRecordPro.java - * Editor: Waue Chen - * From : NCHC. Taiwn - * Last Update Date: 07/02/2008 - * Upgrade to 0.17 - */ - -/** - * 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 { - public void reduce(LongWritable key, Iterator values, - OutputCollector 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); - // 0.16 -// conf.setInputPath(text_path); - Convert.setInputPath(conf, 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); - FileSystem.get(conf).delete(text_path,true); - - setup.deleteFile(conf_tmp); - } -} Index: mple/WordCount.java =================================================================== --- /sample/WordCount.java (revision 22) +++ (revision ) @@ -1,132 +1,0 @@ -/** - * Program: WordCount.java - * Editor: Waue Chen - * From : NCHC. Taiwn - * Last Update Date: 07/02/2008 - * Upgrade to 0.17 - */ - -/** - * Purpose : - * Store the result of WordCount.java from Hbase to Hadoop file system - * - * HowToUse : - * Make sure Hadoop file system is running correctly. - * Put text file on the directory "/local_src/input" - * You can use the instruction to upload "/local_src/input" to HDFS input dir - * $ bin/hadoop dfs -put /local_src/input input - * Then modify the $filepath parameter in construtor to be correct and run this code. - * - * - * Check Result: - * inspect http://localhost:50070 by web explorer - */ -package tw.org.nchc.code; - -import java.io.IOException; -import java.util.Iterator; -import java.util.StringTokenizer; - -import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.fs.Path; -import org.apache.hadoop.io.IntWritable; -import org.apache.hadoop.io.LongWritable; -import org.apache.hadoop.io.Text; -import org.apache.hadoop.mapred.JobClient; -import org.apache.hadoop.mapred.JobConf; -import org.apache.hadoop.mapred.MapReduceBase; -import org.apache.hadoop.mapred.Mapper; -import org.apache.hadoop.mapred.OutputCollector; -import org.apache.hadoop.mapred.Reducer; -import org.apache.hadoop.mapred.Reporter; - -public class WordCount { - private String filepath; - - private String outputPath; - - public WordCount() { - filepath = "/user/waue/input/"; - outputPath = "counts1"; - } - - public WordCount(String path, String output) { - filepath = path; - outputPath = output; - } - - // mapper: emits (token, 1) for every word occurrence - private static class MapClass extends MapReduceBase implements - Mapper { - - // reuse objects to save overhead of object creation - private final static IntWritable one = new IntWritable(1); - - private Text word = new Text(); - - public void map(LongWritable key, Text value, - OutputCollector output, Reporter reporter) - throws IOException { - String line = ((Text) value).toString(); - StringTokenizer itr = new StringTokenizer(line); - while (itr.hasMoreTokens()) { - word.set(itr.nextToken()); - output.collect(word, one); - } - } - } - - // reducer: sums up all the counts - private static class ReduceClass extends MapReduceBase implements - Reducer { - - // reuse objects - private final static IntWritable SumValue = new IntWritable(); - - public void reduce(Text key, Iterator values, - OutputCollector output, Reporter reporter) - throws IOException { - // sum up values - int sum = 0; - while (values.hasNext()) { - sum += values.next().get(); - } - SumValue.set(sum); - output.collect(key, SumValue); - } - } - - /** - * Runs the demo. - */ - public static void main(String[] args) throws IOException { - WordCount wc = new WordCount(); - - int mapTasks = 1; - int reduceTasks = 1; - JobConf conf = new JobConf(WordCount.class); - conf.setJobName("wordcount"); - - conf.setNumMapTasks(mapTasks); - conf.setNumReduceTasks(reduceTasks); - // 0.16 - // conf.setInputPath(new Path(wc.filepath)); - Convert.setInputPath(conf, new Path(wc.filepath)); - conf.setOutputKeyClass(Text.class); - conf.setOutputValueClass(IntWritable.class); - // 0.16 - // conf.setOutputPath(new Path(wc.outputPath)); - Convert.setOutputPath(conf, new Path(wc.outputPath)); - - conf.setMapperClass(MapClass.class); - conf.setCombinerClass(ReduceClass.class); - conf.setReducerClass(ReduceClass.class); - - // Delete the output directory if it exists already - Path outputDir = new Path(wc.outputPath); - // 0.16 - FileSystem.get(conf).delete(outputDir,true); - - JobClient.runJob(conf); - } -} Index: mple/WordCountFromHBase.java =================================================================== --- /sample/WordCountFromHBase.java (revision 22) +++ (revision ) @@ -1,177 +1,0 @@ -/** - * Program: WordCountFromHBase.java - * Editor: Waue Chen - * From : NCHC. Taiwn - * Last Update Date: 07/02/2008 - * Upgrade to 0.17 - */ - -/** - * Purpose : - * Word counting from Hbase then store result in Hadoop file system - * - * HowToUse : - * Make sure Hadoop file system are running and HBase has correct data. - * Suggest to run WordCountIntoHBase first. - * finally, modify these setup parameters and run. - * - * Check Result: - * - * inspect http://localhost:50070 by web explorer - */ - -package tw.org.nchc.code; - -import java.io.IOException; -import java.util.Iterator; -import java.util.StringTokenizer; - -import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.fs.Path; -import org.apache.hadoop.hbase.HStoreKey; -import org.apache.hadoop.hbase.io.ImmutableBytesWritable; -import org.apache.hadoop.hbase.mapred.TableInputFormat; -import org.apache.hadoop.hbase.mapred.TableMap; -import org.apache.hadoop.io.IntWritable; -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.MapReduceBase; -import org.apache.hadoop.mapred.OutputCollector; -import org.apache.hadoop.mapred.Reducer; -import org.apache.hadoop.mapred.Reporter; -@SuppressWarnings("unused") - -public class WordCountFromHBase { - /* setup parameters */ - // set the output path - static String outputPath = "counts2"; - - // org.apache.hadoop.hbase.mapred.TableMap \ - // TableMap \ - // Scan an HBase table to sort by a specified sort column. \ - // If the column does not exist, the record is not passed to Reduce.; - private static class MapClass extends TableMap { - - // set one as (IntWritable)1 - private final static IntWritable one = new IntWritable(1); - // set column - private final static Text textcol = new Text(WordCountIntoHBase.colstr); - private Text word = new Text(); - // TableMap is a interface, map is a abstract method. now, we should \ - // inprement map() at here, format is : \ - // map(HStoreKey key, MapWritable value, \ - // OutputCollector output, Reporter reporter) ; - // Call a user defined function on a single HBase record, \ - // represented by a key and its associated record value. ; - public void map(HStoreKey key, MapWritable cols, - OutputCollector output, Reporter reporter) - throws IOException { - // - // The first get() is : Writable <- get(Object key) \ - // get in interface Map ; - // Use ImmutableBytesWritable to downcast Writable \ - // The second get() is : byte[] <- get() \ - // Get the data from the BytesWritable. ; - // Text.decode is parse UTF-8 code to a String ; - // per "line" is per row data in HTable - String line = Text.decode( ((ImmutableBytesWritable) cols.get(textcol) ) - .get() ); - - //let us know what is "line" - /* - RandomAccessFile raf = - new RandomAccessFile("/home/waue/mr-result.txt","rw"); - raf.seek(raf.length()); // move pointer to end - raf.write(("\n"+line).getBytes()); - raf.close(); - *///end - // the result is the contents of merged files " - - //StringTokenizer will divide a line into a word - StringTokenizer itr = new StringTokenizer(line); - // set every word as one - while (itr.hasMoreTokens()) { - // nextToken will return this value in String and point to next \ - // Text.set() = Set to contain the contents of a string. - word.set(itr.nextToken()); - // OutputCollector.collect = collect(K key, V value) \ - // Adds a key/value pair to the output. - output.collect(word, one); - } - } - } - - // reducer: sums up all the counts - private static class ReduceClass extends MapReduceBase implements - Reducer { - - // reuse objects - private final static IntWritable SumValue = new IntWritable(); - - // this sample's reduce() format is the same as map() \ - // reduce is a method waiting for implement \ - // four type in this sample is (Text , Iterator, \ - // OutputCollector , Reporter ) ; - public void reduce(Text key, Iterator values, - OutputCollector output, Reporter reporter) - throws IOException { - // sum up value - int sum = 0; - // "key" is word , "value" is sum - // why values.hasNext(), not key.hasNext() - while (values.hasNext()) { - // next() will return this value and pointer to next event \ - // IntWritable.get() will transfer IntWritable to Int - sum += values.next().get(); - } - // IntWritable.set(int) will transfer Int to IntWritable - SumValue.set(sum); - // hense we set outputPath in main, the output.collect will put - // data in Hadoop - output.collect(key, SumValue); - } - } - - private WordCountFromHBase() { - } - - /** - * Runs the demo. - */ - public static void main(String[] args) throws IOException { - - - int mapTasks = 1; - int reduceTasks = 1; - // initialize job; - JobConf conf = new JobConf(WordCountFromHBase.class); - // TableMap.initJob will build HBase code \ - // "org.apache.hadoop.hbase.mapred.TableMap".initJob \ - // (Table_name,column_string,Which_class_will_use,job_configure); - TableMap.initJob(WordCountIntoHBase.Table_Name, - WordCountIntoHBase.colstr, MapClass.class, conf); - conf.setJobName(WordCountIntoHBase.Table_Name + "store"); - conf.setNumMapTasks(mapTasks); - conf.setNumReduceTasks(reduceTasks); - - //Set the key class for the job output data. - conf.setOutputKeyClass(Text.class); - //Set the value class for job outputs. - conf.setOutputValueClass(IntWritable.class); - // MapperClass,CombinerClass,ReducerClass are essential - conf.setMapperClass(MapClass.class); - conf.setCombinerClass(ReduceClass.class); - conf.setReducerClass(ReduceClass.class); - // input is Hbase format => TableInputFormat - conf.setInputFormat(TableInputFormat.class); - // 0.16 -// conf.setOutputPath(new Path(outputPath)); - Convert.setOutputPath(conf, new Path(outputPath)); -// delete the old path with the same name - FileSystem.get(conf).delete(new Path(outputPath),true); - JobClient.runJob(conf); - } -} Index: mple/WordCountIntoHBase.java =================================================================== --- /sample/WordCountIntoHBase.java (revision 22) +++ (revision ) @@ -1,118 +1,0 @@ -/** - * Program: WordCountIntoHBase.java - * Editor: Waue Chen - * From : NCHC. Taiwn - * Last Update Date: 07/02/2008 - * Upgrade to 0.17 - */ - -/** - * Purpose : - * Store every line from $Input_Path to HBase - * - * HowToUse : - * Make sure Hadoop file system and HBase are running correctly. - * Use Hadoop instruction to add input-text-files to $Input_Path. - * ($ bin/hadoop dfs -put local_dir hdfs_dir) - * Then run the program with BuildHTable.java after \ - * modifying these setup parameters. - * - * Check Result : - * View the result by hbase instruction (hql> select * from $Table_Name). - * Or run WordCountFromHBase.java then inspect http://localhost:60070 by web explorer; - */ - -package tw.org.nchc.code; - -import java.io.IOException; -import java.util.Iterator; - -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 WordCountIntoHBase { - - /* setup parameters */ - // $Input_Path. Please make sure the path is correct and contains input - // files - static final String Input_Path = "/user/waue/simple"; - - // Hbase table name, the program will create it - static final String Table_Name = "word_count5"; - - // column name, the program will create it - static final String colstr = "word:text"; - - // constructor - private WordCountIntoHBase() { - } - - private static class ReduceClass extends TableReduce { - // set (column_family:column_qualify) - private static final Text col = new Text(WordCountIntoHBase.colstr); - - // this map holds the columns per row - private MapWritable map = new MapWritable(); - - public void reduce(LongWritable key, Iterator values, - OutputCollector output, Reporter reporter) - throws IOException { - // contents must be ImmutableBytesWritable - ImmutableBytesWritable bytes = new ImmutableBytesWritable(values - .next().getBytes()); - map.clear(); - // write data - map.put(col, bytes); - // add the row with the key as the row id - output.collect(new Text(key.toString()), map); - } - } - - /** - * Runs the demo. - */ - public static void main(String[] args) throws IOException { - // parse colstr to split column family and column qualify - String tmp[] = colstr.split(":"); - String Column_Family = tmp[0] + ":"; - String CF[] = { Column_Family }; - // check whether create table or not , we don't admit \ - // the same name but different structure - BuildHTable build_table = new BuildHTable(Table_Name, CF); - 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 !"); - } - int mapTasks = 1; - int reduceTasks = 1; - JobConf conf = new JobConf(WordCountIntoHBase.class); - conf.setJobName(Table_Name); - - // must initialize the TableReduce before running job - TableReduce.initJob(Table_Name, ReduceClass.class, conf); - conf.setNumMapTasks(mapTasks); - conf.setNumReduceTasks(reduceTasks); - // 0.16 - // conf.setInputPath(new Path(Input_Path)); - Convert.setInputPath(conf, new Path(Input_Path)); - conf.setMapperClass(IdentityMapper.class); - conf.setCombinerClass(IdentityReducer.class); - conf.setReducerClass(ReduceClass.class); - - JobClient.runJob(conf); - } -} Index: mple/hadoop-0.16/Convert.java =================================================================== --- /sample/hadoop-0.16/Convert.java (revision 22) +++ (revision ) @@ -1,33 +1,0 @@ -package tw.org.nchc.code; -import java.io.IOException; - -import org.apache.hadoop.fs.FileStatus; -import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.fs.Path; -import org.apache.hadoop.mapred.FileInputFormat; -import org.apache.hadoop.mapred.FileOutputFormat; -import org.apache.hadoop.mapred.JobConf; - -public class Convert -{ - static public Path[] listPaths(FileSystem fsm,Path path) throws IOException - { - FileStatus[] fss = fsm.listStatus(path); - int length = fss.length; - Path[] pi = new Path[length]; - for (int i=0 ; i< length; i++) - { - pi[i] = fss[i].getPath(); - } - return pi; - } - static public void setInputPath(JobConf conf, Path path){ - FileInputFormat.setInputPaths(conf, path); - } - static public void addInputPath(JobConf conf, Path path){ - FileInputFormat.addInputPath(conf, path); - } - static public void setOutputPath(JobConf conf,Path path){ - FileOutputFormat.setOutputPath(conf, path); - } -} Index: /sample/hadoop-0.16/tw/org/nchc/code/BuildHTable.java =================================================================== --- /sample/hadoop-0.16/tw/org/nchc/code/BuildHTable.java (revision 23) +++ /sample/hadoop-0.16/tw/org/nchc/code/BuildHTable.java (revision 23) @@ -0,0 +1,70 @@ +/** + * Program: BuildHTable.java + * Editor: Waue Chen + * From : NCHC. Taiwn + * Last Update Date: 06/10/2008 + */ + +package tw.org.nchc.code; +import java.io.IOException; +import org.apache.hadoop.hbase.HBaseAdmin; +import org.apache.hadoop.hbase.HBaseConfiguration; +import org.apache.hadoop.hbase.HColumnDescriptor; +import org.apache.hadoop.hbase.HTableDescriptor; +import org.apache.hadoop.io.Text; + +public class BuildHTable { + private String table_name ; + private String[] Column_Family ; + HBaseConfiguration conf = new HBaseConfiguration(); + HBaseAdmin admin = new HBaseAdmin(conf); + public BuildHTable(String table,String[] CF) throws IOException{ + table_name = table; + Column_Family = CF; + } + + + public boolean checkTableExist(String tname)throws IOException{ + if(! admin.tableExists(new Text(tname))){ + return false; + } + return true; + } + // create Hbase table , success = 1 ; failse = 0; Table_exist = 2; + public boolean createTable() throws IOException{ + // check whether Table name exite or not + if(! checkTableExist(table_name)){ + System.out.println("HTable : " + table_name + " creating ... please wait"); + HTableDescriptor tableDesc = new HTableDescriptor(table_name); + for(int i =0; i" +Column_Family[i]); + } + //add column family + tableDesc.addFamily(new HColumnDescriptor(Column_Family[i])); + } + admin.createTable(tableDesc); + } else { + return false; + } + return true; + } + public static void main(String[] args) throws IOException { + + // setup Table name + String Table_Name = "test_create_table2"; + // setup Column Family + String[] Column_Family = {"http:","url:","referrer:"}; + + BuildHTable bt = new BuildHTable( Table_Name , Column_Family); + boolean ret = bt.createTable(); + if(ret == true){ + System.out.println("Create Table \"" +Table_Name +" \" Compelte !!!"); + }else { + System.out.println("Table Name \"" +Table_Name +" \" exit!!!"); + } + } +} Index: /sample/hadoop-0.16/tw/org/nchc/code/Convert.java =================================================================== --- /sample/hadoop-0.16/tw/org/nchc/code/Convert.java (revision 23) +++ /sample/hadoop-0.16/tw/org/nchc/code/Convert.java (revision 23) @@ -0,0 +1,33 @@ +package tw.org.nchc.code; +import java.io.IOException; + +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.mapred.FileInputFormat; +import org.apache.hadoop.mapred.FileOutputFormat; +import org.apache.hadoop.mapred.JobConf; + +public class Convert +{ + static public Path[] listPaths(FileSystem fsm,Path path) throws IOException + { + FileStatus[] fss = fsm.listStatus(path); + int length = fss.length; + Path[] pi = new Path[length]; + for (int i=0 ; i< length; i++) + { + pi[i] = fss[i].getPath(); + } + return pi; + } + static public void setInputPath(JobConf conf, Path path){ + FileInputFormat.setInputPaths(conf, path); + } + static public void addInputPath(JobConf conf, Path path){ + FileInputFormat.addInputPath(conf, path); + } + static public void setOutputPath(JobConf conf,Path path){ + FileOutputFormat.setOutputPath(conf, path); + } +} Index: /sample/hadoop-0.16/tw/org/nchc/code/HBaseClient.java =================================================================== --- /sample/hadoop-0.16/tw/org/nchc/code/HBaseClient.java (revision 23) +++ /sample/hadoop-0.16/tw/org/nchc/code/HBaseClient.java (revision 23) @@ -0,0 +1,152 @@ +/** + * Program: HBaseClient.java + * Editor: Waue Chen + * From : NCHC. Taiwn + * Last Update Date: 06/10/2008 + */ + +package tw.org.nchc.code; + +import java.io.IOException; +import java.util.SortedMap; +import java.util.TreeMap; + +import org.apache.hadoop.hbase.HBaseConfiguration; +import org.apache.hadoop.hbase.HScannerInterface; +import org.apache.hadoop.hbase.HStoreKey; +import org.apache.hadoop.hbase.HTable; +import org.apache.hadoop.io.Text; + +/** + * Demo that illustrates the HBase client API. + * The demo program will insert values to " Column Family: Column qualifier" and then print these. + * pre-do : create a hbase table "test_table" with (CF:CI) which (" column family ", " column qualifier ") + * 1. $ bin/hbase shell + * 2. > create table test_table("CF"); + * ok ! we can test it + * 3. > insert into test_table("CF:CI") values=("Hellow World") where row = "1"; + * 4. > select * from test_table; + +08/06/03 16:16:36 INFO hbase.HTable: Creating scanner over test_table starting at key ++---------+-----------+-----------+ +| Row | Column | Cell | ++---------+-----------+-----------+ +| 1 | CF:CI | Hellow World | ++---------+-----------+-----------+ +1 row(s) in set. (0.24 sec) + + * on the structure , "Row" means Row ID which is a key to describe a column; + * Column means the database structure in test_table, + * Column Family , "CF", should be defined while creating table. + * Column qualifier , "CI" , can be added dynamically. + * Cell is the value of CF:CI + * + * that's the structure; then the demo program will show you in console as below : + * +Illustration of adding data... +Writing row = 0, col 'CF:CI' = Hellow0 +Writing row = 1, col 'CF:CI' = Hellow1 +Writing row = 2, col 'CF:CI' = Hellow2 +Writing row = 3, col 'CF:CI' = Hellow3 +Writing row = 4, col 'CF:CI' = Hellow4 +Writing row = 5, col 'CF:CI' = Hellow5 +Writing row = 6, col 'CF:CI' = Hellow6 +Writing row = 7, col 'CF:CI' = Hellow7 +Writing row = 8, col 'CF:CI' = Hellow8 +Writing row = 9, col 'CF:CI' = Hellow9 + +Illustration of querying... +row = 1, 'CF : CI ' = Hellow1 + +Illustration of scanning... +08/06/03 16:47:51 INFO hbase.HTable: Creating scanner over test_table starting at key +row = 0//9223372036854775807, col 'CF:CI' = Hellow0 +row = 1//9223372036854775807, col 'CF:CI' = Hellow1 +row = 2//9223372036854775807, col 'CF:CI' = Hellow2 +row = 3//9223372036854775807, col 'CF:CI' = Hellow3 +row = 4//9223372036854775807, col 'CF:CI' = Hellow4 +row = 5//9223372036854775807, col 'CF:CI' = Hellow5 +row = 6//9223372036854775807, col 'CF:CI' = Hellow6 +row = 7//9223372036854775807, col 'CF:CI' = Hellow7 +row = 8//9223372036854775807, col 'CF:CI' = Hellow8 +row = 9//9223372036854775807, col 'CF:CI' = Hellow9 + + + * + */ +public class HBaseClient { + + public static void main(String[] args) throws IOException { + + // Open the "test_table" table. If it hasn't been in Hbase, you should create. + HBaseConfiguration conf = new HBaseConfiguration(); + HTable table = new HTable(conf, new Text("test_table")); + + System.out.println("Illustration of adding data..."); + + // create column formed (Column Family:Column qualifier) + Text column = new Text("CF:CI"); + + // create row_id + Text row_id = new Text(); + + // demo 1 : Insert ten demo values + for (int i = 0; i < 10; i++) { + + // give row_id value + row_id.set(new Integer(i).toString()); + + // let "indicate_id" indicate the column which row = row_id + long indicate_id= table.startUpdate(row_id); + + //val = value of CF:CI where row_id = i + Text val = new Text("Hellow" + i); + + // put "val" to "column" from "table" where "row_id" + // the same as : + // hql> INSERT INTO table( column ) VALUES=( val) WHERE ROW = row_id ; + table.put(indicate_id, column, val.getBytes()); + table.commit(indicate_id); + + System.out.println("Writing row = " + row_id + ", col '" + column + + "' = " + val); + } + + // demo 2 : print column value only row = 1 ; + System.out.println("\n Querying row = 1"); + + // Get a single value for the specified row and column + // byte[] = HTable.get(Text row, Text column) + + String s = Text.decode(table.get(new Text("1"),new Text("CF:CI"))); + // if change as + // String s = (table.get(new Text("1"),new Text("CF:CI"))).toString(); + // will get chaos code " [B@1f14ceb" + System.out.println("row = 1, 'CF : CI ' = " + s); + + // demo 3 : Print the all contents of this table + System.out.println("\nIllustration of scanning..."); + + // we only want one column, but you can specify multiple columns to + // fetch at once + Text[] cols = { column }; + + // Use HScannerInterface to crawl table + HScannerInterface scanner = table.obtainScanner(cols, new Text()); + + // column values are stored in a Map + SortedMap values = new TreeMap(); + HStoreKey currentKey = new HStoreKey(); + while (scanner.next(currentKey, values)) { + // decode the stored byte[] back into a String + String val = Text.decode(values.get(column)); + System.out.println("row = " + currentKey + ", col '" + column + "' = " + + val); + } + + // remember to close scanner when done + scanner.close(); + + } + +} Index: /sample/hadoop-0.16/tw/org/nchc/code/HBaseRecord.java =================================================================== --- /sample/hadoop-0.16/tw/org/nchc/code/HBaseRecord.java (revision 23) +++ /sample/hadoop-0.16/tw/org/nchc/code/HBaseRecord.java (revision 23) @@ -0,0 +1,174 @@ +/** + * Program: HBaseRecord.java + * Editor: Waue Chen + * From : NCHC. Taiwn + * Last Update Date: 07/02/2008 + * Upgrade to 0.17 + */ + +/** + * Purpose : + * Parse your record and then store in HBase. + * + * 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 | + ---------------- + * 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) + */ + +package tw.org.nchc.code; + +import java.io.IOException; +import java.util.Iterator; + +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 HBaseRecord { + + /* Denify parameter */ + // one column family: person; three column qualifier: name,locate,years + static private String baseId1 = "person:name"; + + static private String baseId2 = "person:locate"; + + static private String baseId3 = "person:years"; + + // split character + static private String sp = ":"; + + // file path in hadoop file system (not phisical file system) + String file_path = "/user/waue/t1"; + + // Hbase table name + String table_name = "t1_table"; + + // setup MapTask and Reduce Task + int mapTasks = 1; + + int reduceTasks = 1; + + private static class ReduceClass extends TableReduce { + + // Column id is created dymanically, + private static final Text col_name = new Text(baseId1); + + private static final Text col_local = new Text(baseId2); + + private static final Text col_year = new Text(baseId3); + + // this map holds the columns per row + private MapWritable map = new MapWritable(); + + // on this sample, map is nonuse, we use reduce to handle + public void reduce(LongWritable key, Iterator values, + OutputCollector output, Reporter reporter) + throws IOException { + + // values.next().getByte() can get value and transfer to byte form, + // there is an other way that let decode() + // to substitude getByte() + String stro = new String(values.next().getBytes()); + String str[] = stro.split(sp); + byte b_local[] = str[0].getBytes(); + byte b_name[] = str[1].getBytes(); + byte b_year[] = str[2].getBytes(); + + // contents must be ImmutableBytesWritable + ImmutableBytesWritable w_local = new ImmutableBytesWritable(b_local); + ImmutableBytesWritable w_name = new ImmutableBytesWritable(b_name); + ImmutableBytesWritable w_year = new ImmutableBytesWritable(b_year); + + // populate the current row + map.clear(); + map.put(col_name, w_local); + map.put(col_local, w_name); + map.put(col_year, w_year); + + // add the row with the key as the row id + output.collect(new Text(key.toString()), map); + } + } + + private HBaseRecord() { + } + + /** + * Runs the demo. + */ + public static void main(String[] args) throws IOException { + // which path of input files in Hadoop file system + + HBaseRecord setup = new HBaseRecord(); + JobConf conf = new JobConf(HBaseRecord.class); + + // Job name; you can modify to any you like + conf.setJobName("NCHC_PersonDataBase"); + + // Hbase table name must be correct , in our profile is t1_table + TableReduce.initJob(setup.table_name, ReduceClass.class, conf); + + // below are map-reduce profile + conf.setNumMapTasks(setup.mapTasks); + conf.setNumReduceTasks(setup.reduceTasks); + + // 0.16 + // conf.setInputPath(new Path(setup.file_path)); + Convert.setInputPath(conf, new Path(setup.file_path)); + + conf.setMapperClass(IdentityMapper.class); + conf.setCombinerClass(IdentityReducer.class); + conf.setReducerClass(ReduceClass.class); + JobClient.runJob(conf); + } +} Index: /sample/hadoop-0.16/tw/org/nchc/code/HBaseRecord2.java =================================================================== --- /sample/hadoop-0.16/tw/org/nchc/code/HBaseRecord2.java (revision 23) +++ /sample/hadoop-0.16/tw/org/nchc/code/HBaseRecord2.java (revision 23) @@ -0,0 +1,162 @@ +/** + * Program: HBaseRecord.java + * Editor: Waue Chen + * From : NCHC. Taiwn + * Last Update Date: 07/01/2008 + * Upgrade to 0.17 + */ + +/** + * Purpose : + * Parse your record and then store in HBase. + * + * 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 + ---------------- + * 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) + */ + + + + +package tw.org.nchc.code; + +import java.io.IOException; +import java.util.Iterator; + +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 HBaseRecord2 { + + /* Denify parameter */ + static String[] bf = {"person:name","person:local","person:birthyear"}; + // file path in hadoop file system (not phisical file system) + String file_path = "/user/waue/t1/test.txt"; + // Hbase table name + String table_name = "testtable"; + + + // setup MapTask and Reduce Task + int mapTasks = 1; + int reduceTasks = 1; + + private static class ReduceClass extends TableReduce { + + + + // on this sample, map is nonuse, we use reduce to handle + public void reduce(LongWritable key, Iterator values, + OutputCollector output, Reporter reporter) + throws IOException { + // 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(":"); + + int length = bf.length; + + // 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[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); + } + } + + private HBaseRecord2() { + } + + /** + * Runs the demo. + */ + public static void main(String[] args) throws IOException { + + + HBaseRecord2 setup = new HBaseRecord2(); + String[] tmp = bf[0].split(":"); + String[] CF = {tmp[0]}; + BuildHTable build_table = new BuildHTable(setup.table_name, CF); + if (!build_table.checkTableExist(setup.table_name)) { + if (!build_table.createTable()) { + System.out.println("create table error !"); + } + } else { + System.out.println("Table \"" + setup.table_name + + "\" has already existed !"); + } + + JobConf conf = new JobConf(HBaseRecord2.class); + + //Job name; you can modify to any you like + conf.setJobName("PersonDataBase"); + + // Hbase table name must be correct , in our profile is t1_table + TableReduce.initJob(setup.table_name, ReduceClass.class, conf); + + // below are map-reduce profile + conf.setNumMapTasks(setup.mapTasks); + conf.setNumReduceTasks(setup.reduceTasks); + // 0.16 +// conf.setInputPath(new Path(setup.file_path)); + Convert.setInputPath(conf, new Path(setup.file_path)); + conf.setMapperClass(IdentityMapper.class); + conf.setCombinerClass(IdentityReducer.class); + conf.setReducerClass(ReduceClass.class); + JobClient.runJob(conf); + } +} Index: /sample/hadoop-0.16/tw/org/nchc/code/HBaseRecordPro.java =================================================================== --- /sample/hadoop-0.16/tw/org/nchc/code/HBaseRecordPro.java (revision 23) +++ /sample/hadoop-0.16/tw/org/nchc/code/HBaseRecordPro.java (revision 23) @@ -0,0 +1,247 @@ +/** + * Program: HBaseRecordPro.java + * Editor: Waue Chen + * From : NCHC. Taiwn + * Last Update Date: 07/02/2008 + * Upgrade to 0.17 + */ + +/** + * 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 { + public void reduce(LongWritable key, Iterator values, + OutputCollector 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); + // 0.16 +// conf.setInputPath(text_path); + Convert.setInputPath(conf, 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); + FileSystem.get(conf).delete(text_path,true); + + setup.deleteFile(conf_tmp); + } +} Index: /sample/hadoop-0.16/tw/org/nchc/code/WordCount.java =================================================================== --- /sample/hadoop-0.16/tw/org/nchc/code/WordCount.java (revision 23) +++ /sample/hadoop-0.16/tw/org/nchc/code/WordCount.java (revision 23) @@ -0,0 +1,132 @@ +/** + * Program: WordCount.java + * Editor: Waue Chen + * From : NCHC. Taiwn + * Last Update Date: 07/02/2008 + * Upgrade to 0.17 + */ + +/** + * Purpose : + * Store the result of WordCount.java from Hbase to Hadoop file system + * + * HowToUse : + * Make sure Hadoop file system is running correctly. + * Put text file on the directory "/local_src/input" + * You can use the instruction to upload "/local_src/input" to HDFS input dir + * $ bin/hadoop dfs -put /local_src/input input + * Then modify the $filepath parameter in construtor to be correct and run this code. + * + * + * Check Result: + * inspect http://localhost:50070 by web explorer + */ +package tw.org.nchc.code; + +import java.io.IOException; +import java.util.Iterator; +import java.util.StringTokenizer; + +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.IntWritable; +import org.apache.hadoop.io.LongWritable; +import org.apache.hadoop.io.Text; +import org.apache.hadoop.mapred.JobClient; +import org.apache.hadoop.mapred.JobConf; +import org.apache.hadoop.mapred.MapReduceBase; +import org.apache.hadoop.mapred.Mapper; +import org.apache.hadoop.mapred.OutputCollector; +import org.apache.hadoop.mapred.Reducer; +import org.apache.hadoop.mapred.Reporter; + +public class WordCount { + private String filepath; + + private String outputPath; + + public WordCount() { + filepath = "/user/waue/input/"; + outputPath = "counts1"; + } + + public WordCount(String path, String output) { + filepath = path; + outputPath = output; + } + + // mapper: emits (token, 1) for every word occurrence + private static class MapClass extends MapReduceBase implements + Mapper { + + // reuse objects to save overhead of object creation + private final static IntWritable one = new IntWritable(1); + + private Text word = new Text(); + + public void map(LongWritable key, Text value, + OutputCollector output, Reporter reporter) + throws IOException { + String line = ((Text) value).toString(); + StringTokenizer itr = new StringTokenizer(line); + while (itr.hasMoreTokens()) { + word.set(itr.nextToken()); + output.collect(word, one); + } + } + } + + // reducer: sums up all the counts + private static class ReduceClass extends MapReduceBase implements + Reducer { + + // reuse objects + private final static IntWritable SumValue = new IntWritable(); + + public void reduce(Text key, Iterator values, + OutputCollector output, Reporter reporter) + throws IOException { + // sum up values + int sum = 0; + while (values.hasNext()) { + sum += values.next().get(); + } + SumValue.set(sum); + output.collect(key, SumValue); + } + } + + /** + * Runs the demo. + */ + public static void main(String[] args) throws IOException { + WordCount wc = new WordCount(); + + int mapTasks = 1; + int reduceTasks = 1; + JobConf conf = new JobConf(WordCount.class); + conf.setJobName("wordcount"); + + conf.setNumMapTasks(mapTasks); + conf.setNumReduceTasks(reduceTasks); + // 0.16 + // conf.setInputPath(new Path(wc.filepath)); + Convert.setInputPath(conf, new Path(wc.filepath)); + conf.setOutputKeyClass(Text.class); + conf.setOutputValueClass(IntWritable.class); + // 0.16 + // conf.setOutputPath(new Path(wc.outputPath)); + Convert.setOutputPath(conf, new Path(wc.outputPath)); + + conf.setMapperClass(MapClass.class); + conf.setCombinerClass(ReduceClass.class); + conf.setReducerClass(ReduceClass.class); + + // Delete the output directory if it exists already + Path outputDir = new Path(wc.outputPath); + // 0.16 + FileSystem.get(conf).delete(outputDir,true); + + JobClient.runJob(conf); + } +} Index: /sample/hadoop-0.16/tw/org/nchc/code/WordCountFromHBase.java =================================================================== --- /sample/hadoop-0.16/tw/org/nchc/code/WordCountFromHBase.java (revision 23) +++ /sample/hadoop-0.16/tw/org/nchc/code/WordCountFromHBase.java (revision 23) @@ -0,0 +1,177 @@ +/** + * Program: WordCountFromHBase.java + * Editor: Waue Chen + * From : NCHC. Taiwn + * Last Update Date: 07/02/2008 + * Upgrade to 0.17 + */ + +/** + * Purpose : + * Word counting from Hbase then store result in Hadoop file system + * + * HowToUse : + * Make sure Hadoop file system are running and HBase has correct data. + * Suggest to run WordCountIntoHBase first. + * finally, modify these setup parameters and run. + * + * Check Result: + * + * inspect http://localhost:50070 by web explorer + */ + +package tw.org.nchc.code; + +import java.io.IOException; +import java.util.Iterator; +import java.util.StringTokenizer; + +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.HStoreKey; +import org.apache.hadoop.hbase.io.ImmutableBytesWritable; +import org.apache.hadoop.hbase.mapred.TableInputFormat; +import org.apache.hadoop.hbase.mapred.TableMap; +import org.apache.hadoop.io.IntWritable; +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.MapReduceBase; +import org.apache.hadoop.mapred.OutputCollector; +import org.apache.hadoop.mapred.Reducer; +import org.apache.hadoop.mapred.Reporter; +@SuppressWarnings("unused") + +public class WordCountFromHBase { + /* setup parameters */ + // set the output path + static String outputPath = "counts2"; + + // org.apache.hadoop.hbase.mapred.TableMap \ + // TableMap \ + // Scan an HBase table to sort by a specified sort column. \ + // If the column does not exist, the record is not passed to Reduce.; + private static class MapClass extends TableMap { + + // set one as (IntWritable)1 + private final static IntWritable one = new IntWritable(1); + // set column + private final static Text textcol = new Text(WordCountIntoHBase.colstr); + private Text word = new Text(); + // TableMap is a interface, map is a abstract method. now, we should \ + // inprement map() at here, format is : \ + // map(HStoreKey key, MapWritable value, \ + // OutputCollector output, Reporter reporter) ; + // Call a user defined function on a single HBase record, \ + // represented by a key and its associated record value. ; + public void map(HStoreKey key, MapWritable cols, + OutputCollector output, Reporter reporter) + throws IOException { + // + // The first get() is : Writable <- get(Object key) \ + // get in interface Map ; + // Use ImmutableBytesWritable to downcast Writable \ + // The second get() is : byte[] <- get() \ + // Get the data from the BytesWritable. ; + // Text.decode is parse UTF-8 code to a String ; + // per "line" is per row data in HTable + String line = Text.decode( ((ImmutableBytesWritable) cols.get(textcol) ) + .get() ); + + //let us know what is "line" + /* + RandomAccessFile raf = + new RandomAccessFile("/home/waue/mr-result.txt","rw"); + raf.seek(raf.length()); // move pointer to end + raf.write(("\n"+line).getBytes()); + raf.close(); + *///end + // the result is the contents of merged files " + + //StringTokenizer will divide a line into a word + StringTokenizer itr = new StringTokenizer(line); + // set every word as one + while (itr.hasMoreTokens()) { + // nextToken will return this value in String and point to next \ + // Text.set() = Set to contain the contents of a string. + word.set(itr.nextToken()); + // OutputCollector.collect = collect(K key, V value) \ + // Adds a key/value pair to the output. + output.collect(word, one); + } + } + } + + // reducer: sums up all the counts + private static class ReduceClass extends MapReduceBase implements + Reducer { + + // reuse objects + private final static IntWritable SumValue = new IntWritable(); + + // this sample's reduce() format is the same as map() \ + // reduce is a method waiting for implement \ + // four type in this sample is (Text , Iterator, \ + // OutputCollector , Reporter ) ; + public void reduce(Text key, Iterator values, + OutputCollector output, Reporter reporter) + throws IOException { + // sum up value + int sum = 0; + // "key" is word , "value" is sum + // why values.hasNext(), not key.hasNext() + while (values.hasNext()) { + // next() will return this value and pointer to next event \ + // IntWritable.get() will transfer IntWritable to Int + sum += values.next().get(); + } + // IntWritable.set(int) will transfer Int to IntWritable + SumValue.set(sum); + // hense we set outputPath in main, the output.collect will put + // data in Hadoop + output.collect(key, SumValue); + } + } + + private WordCountFromHBase() { + } + + /** + * Runs the demo. + */ + public static void main(String[] args) throws IOException { + + + int mapTasks = 1; + int reduceTasks = 1; + // initialize job; + JobConf conf = new JobConf(WordCountFromHBase.class); + // TableMap.initJob will build HBase code \ + // "org.apache.hadoop.hbase.mapred.TableMap".initJob \ + // (Table_name,column_string,Which_class_will_use,job_configure); + TableMap.initJob(WordCountIntoHBase.Table_Name, + WordCountIntoHBase.colstr, MapClass.class, conf); + conf.setJobName(WordCountIntoHBase.Table_Name + "store"); + conf.setNumMapTasks(mapTasks); + conf.setNumReduceTasks(reduceTasks); + + //Set the key class for the job output data. + conf.setOutputKeyClass(Text.class); + //Set the value class for job outputs. + conf.setOutputValueClass(IntWritable.class); + // MapperClass,CombinerClass,ReducerClass are essential + conf.setMapperClass(MapClass.class); + conf.setCombinerClass(ReduceClass.class); + conf.setReducerClass(ReduceClass.class); + // input is Hbase format => TableInputFormat + conf.setInputFormat(TableInputFormat.class); + // 0.16 +// conf.setOutputPath(new Path(outputPath)); + Convert.setOutputPath(conf, new Path(outputPath)); +// delete the old path with the same name + FileSystem.get(conf).delete(new Path(outputPath),true); + JobClient.runJob(conf); + } +} Index: /sample/hadoop-0.16/tw/org/nchc/code/WordCountIntoHBase.java =================================================================== --- /sample/hadoop-0.16/tw/org/nchc/code/WordCountIntoHBase.java (revision 23) +++ /sample/hadoop-0.16/tw/org/nchc/code/WordCountIntoHBase.java (revision 23) @@ -0,0 +1,118 @@ +/** + * Program: WordCountIntoHBase.java + * Editor: Waue Chen + * From : NCHC. Taiwn + * Last Update Date: 07/02/2008 + * Upgrade to 0.17 + */ + +/** + * Purpose : + * Store every line from $Input_Path to HBase + * + * HowToUse : + * Make sure Hadoop file system and HBase are running correctly. + * Use Hadoop instruction to add input-text-files to $Input_Path. + * ($ bin/hadoop dfs -put local_dir hdfs_dir) + * Then run the program with BuildHTable.java after \ + * modifying these setup parameters. + * + * Check Result : + * View the result by hbase instruction (hql> select * from $Table_Name). + * Or run WordCountFromHBase.java then inspect http://localhost:60070 by web explorer; + */ + +package tw.org.nchc.code; + +import java.io.IOException; +import java.util.Iterator; + +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 WordCountIntoHBase { + + /* setup parameters */ + // $Input_Path. Please make sure the path is correct and contains input + // files + static final String Input_Path = "/user/waue/simple"; + + // Hbase table name, the program will create it + static final String Table_Name = "word_count5"; + + // column name, the program will create it + static final String colstr = "word:text"; + + // constructor + private WordCountIntoHBase() { + } + + private static class ReduceClass extends TableReduce { + // set (column_family:column_qualify) + private static final Text col = new Text(WordCountIntoHBase.colstr); + + // this map holds the columns per row + private MapWritable map = new MapWritable(); + + public void reduce(LongWritable key, Iterator values, + OutputCollector output, Reporter reporter) + throws IOException { + // contents must be ImmutableBytesWritable + ImmutableBytesWritable bytes = new ImmutableBytesWritable(values + .next().getBytes()); + map.clear(); + // write data + map.put(col, bytes); + // add the row with the key as the row id + output.collect(new Text(key.toString()), map); + } + } + + /** + * Runs the demo. + */ + public static void main(String[] args) throws IOException { + // parse colstr to split column family and column qualify + String tmp[] = colstr.split(":"); + String Column_Family = tmp[0] + ":"; + String CF[] = { Column_Family }; + // check whether create table or not , we don't admit \ + // the same name but different structure + BuildHTable build_table = new BuildHTable(Table_Name, CF); + 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 !"); + } + int mapTasks = 1; + int reduceTasks = 1; + JobConf conf = new JobConf(WordCountIntoHBase.class); + conf.setJobName(Table_Name); + + // must initialize the TableReduce before running job + TableReduce.initJob(Table_Name, ReduceClass.class, conf); + conf.setNumMapTasks(mapTasks); + conf.setNumReduceTasks(reduceTasks); + // 0.16 + // conf.setInputPath(new Path(Input_Path)); + Convert.setInputPath(conf, new Path(Input_Path)); + conf.setMapperClass(IdentityMapper.class); + conf.setCombinerClass(IdentityReducer.class); + conf.setReducerClass(ReduceClass.class); + + JobClient.runJob(conf); + } +}