Changes between Version 1 and Version 2 of waue/2010/0204-07

Timestamp:

Feb 3, 2010, 9:54:37 PM (14 years ago)

Author:

waue

Comment:

--

waue/2010/0204-07

@@ -27 +27 @@
 {{{
 #!java
+package tsmc;
 
+// LoadHBaseMapper
+//說明： 
+//      此程式碼將HBase的資料取出來，再將結果塞回hdfs上
+//
+//運算方法：
+//      將此程式運作在hadoop 0.20 平台上，用(參考2)的方法加入hbase參數後，將此程式碼打包成XX.jar 
+//  執行：
+//      ---------------------------
+//      hadoop jar XX.jar LoadHBaseMapper <hdfs_output>
+//      ---------------------------
+//
+//結果：
+// $ hadoop fs -cat <hdfs_output>/part-r-00000
+// ---------------------------
+//      54 30 31        GunLong
+//      54 30 32        Esing
+//      54 30 33        SunDon
+//      54 30 34        StarBucks
+// ---------------------------
+//注意：
+//1.    請注意hbase 上必須要有 table, 並且已經有資料
+//2.    運算完後，程式將執行結果放在你指定 hdfs的<hdfs_output> 內
+//              請注意 沒有 <hdfs_output> 資料夾
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
+import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
+import org.apache.hadoop.hbase.mapreduce.TableMapper;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
+
+public class LoadHBaseMapper {
+        public static class HtMap extends TableMapper<Text, Text> {
+
+                public void map(ImmutableBytesWritable key, Result value,
+                                Context context) throws IOException, InterruptedException {
+
+                        String res = Bytes.toString(value.getValue(Bytes.toBytes("Detail"),
+                                        Bytes.toBytes("Name")));
+
+                        context.write(new Text(key.toString()), new Text(res));
+
+                }
+        }
+
+        // TableReducer<KEYIN,VALUEIN,KEYOUT>
+        // 原本為 TableReducer<Text, IntWritable, NullWritable >
+        // 但在此改成 LongWritable 也可以
+        // 因此證明在此的Class可以很多,org.apache.hadoop.io.* 內有write()的Writable class應該皆可
+        public static class HtReduce extends Reducer<Text, Text, Text, Text> {
+
+                public void reduce(Text key, Iterable<Text> values, Context context)
+                                throws IOException, InterruptedException {
+
+                        String str = new String("");
+                        Text final_key = new Text(key);
+                        Text final_value = new Text();
+                        // 將key值相同的values，透過 && 符號分隔之
+                        for (Text tmp : values) {
+                                str += tmp.toString();
+                        }
+                        final_value.set(str);
+                        context.write(final_key, final_value);
+                }
+        }
+
+        public static void main(String args[]) throws Exception {
+                // debug
+                String[] argv = { "output-lhm" };
+                args = argv;
+                String input = args[0];
+
+                String tablename = "tsmc";
+
+                Configuration conf = new Configuration();
+
+                Job job = new Job(conf, tablename + " hbase data to hdfs");
+                job.setJarByClass(LoadHBaseMapper.class);
+
+                // // 設定哪張hbase的table為輸入
+                conf.set(TableInputFormat.INPUT_TABLE, tablename);// 用此方法會遇到
+                                                                                                                        // java.lang.IllegalArgumentException:
+                                                                                                                        // string cannot be
+                                                                                                                        // null
+                Scan myScan = new Scan("".getBytes(), "12".getBytes());
+                myScan.addColumn("Detail:Name".getBytes());
+
+                // 先用 TableMapReduceUtil 的 initTableMapperJob 與 initTableReducerJob 來做設定
+                TableMapReduceUtil.initTableMapperJob(tablename, myScan, HtMap.class,
+                                Text.class, Text.class, job);
+
+                job.setMapperClass(HtMap.class);
+                job.setReducerClass(HtReduce.class);
+
+                job.setMapOutputKeyClass(Text.class);
+                job.setMapOutputValueClass(Text.class);
+
+                job.setInputFormatClass(TableInputFormat.class);
+                job.setOutputFormatClass(TextOutputFormat.class);
+
+                job.setOutputKeyClass(Text.class);
+                job.setOutputValueClass(Text.class);
+
+                FileOutputFormat.setOutputPath(job, new Path(input));
+
+                System.exit(job.waitForCompletion(true) ? 0 : 1);
+        }
+}
 }}}