= 程式宣告 = * Program: HBaseRecordPro.java * Editor: Waue Chen * From : NCHC. Taiwn * Last Update Date: 07/01/2008 ----------------- = 程式功能 = * Program would parse your record and create Hbase * Then it sets the first line as column qualify * Finally it stores in HBase automatically. = 如何使用 = * Make sure two thing : * 1. source_file must be regular as follow: * first line: qualify1:qualify2:...:qualifyN * other line: records1:records2:...:recordsN * 2. source_file path must be correct. = 原始檔內容 = {{{ name:locate:years waue:taiwan:1981 rock:taiwan:1981 aso:taiwan:1981 jazz:taiwan:1982 }}} = 結果 = {{{ * 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); conf.setInputPath(text_path); conf.setMapperClass(IdentityMapper.class); conf.setCombinerClass(IdentityReducer.class); conf.setReducerClass(ReduceClass.class); JobClient.runJob(conf); // delete tmp file FileSystem.get(conf).delete(text_path); setup.deleteFile(conf_tmp); } } }}}