Changes between Initial Version and Version 1 of waue/2011/0425_Itri3CalculateMR

Timestamp:

Apr 25, 2011, 4:20:45 PM (14 years ago)

Author:

waue

Comment:

--

waue/2011/0425_Itri3CalculateMR

@@ +1 @@
+{{{
+#!java
+
+package itri;
+
+//ITRI serial program number 3 
+//0. after Itri2Count 
+//1. run it by MapReduce
+
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.Put;
+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.mapreduce.TableOutputFormat;
+import org.apache.hadoop.hbase.mapreduce.TableReducer;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.Job;
+
+public class Itri3CalculateMR {
+        public static class HtMap extends TableMapper<Text, Text> {
+
+                public void map(ImmutableBytesWritable key, Result value,
+                                Context context) throws IOException, InterruptedException {
+
+                        String row = Bytes.toString(value.getValue(Bytes.toBytes("Detail"),
+                                        Bytes.toBytes("Locate")));
+
+                        int sum = 0;
+
+                        for (int i = 0; i < 4; i++) {
+                                String v = Bytes.toString(value.getValue(Bytes
+                                                .toBytes("Products"), Bytes.toBytes("P" + (i + 1))));
+                                String c = Bytes.toString(value.getValue(Bytes
+                                                .toBytes("Turnover"), Bytes.toBytes("P" + (i + 1))));
+                                if (v != null ) {
+                                        if(c == null) c="0";
+                                        System.err.println("p=" + v);
+                                        System.err.println("c=" + c);
+
+                                        sum += Integer.parseInt(v) * Integer.parseInt(c);
+                                        System.err.println("T" + row + ":" + "p[" + i + "]*" + "c["
+                                                        + i + "] => " + v + "*" + c + "+="
+                                                        + (sum));
+                                        
+                                }
+
+                        }
+
+                        context.write(new Text("T" + row), new Text(String.valueOf(sum)));
+
+                }
+        }
+
+
+        public static class HtReduce extends TableReducer<Text, Text, Text> {
+
+                public void reduce(Text key, Iterable<Text> values, Context context)
+                                throws IOException, InterruptedException {
+                        String sum = "";
+                        for (Text i : values) {
+                                sum += i.toString();
+                        }
+
+                        Put put = new Put(Bytes.toBytes(key.toString()));
+
+                        put.add(Bytes.toBytes("Turnover"), Bytes.toBytes("Sum"), Bytes
+                                        .toBytes(sum));
+
+                        context.write(new Text(), put);
+
+                }
+        }
+
+        public static void main(String args[]) throws Exception {
+
+                String tablename = "itri";
+
+                Scan myScan = new Scan();
+                myScan.addColumn("Detail:Locate".getBytes());
+                myScan.addColumn("Products:P1".getBytes());
+                myScan.addColumn("Products:P2".getBytes());
+                myScan.addColumn("Products:P3".getBytes());
+                myScan.addColumn("Products:P4".getBytes());
+                myScan.addColumn("Turnover:P1".getBytes());
+                myScan.addColumn("Turnover:P2".getBytes());
+                myScan.addColumn("Turnover:P3".getBytes());
+                myScan.addColumn("Turnover:P4".getBytes());
+
+                Configuration conf = new Configuration();
+
+                Job job = new Job(conf, "Calculating ");
+
+                job.setJarByClass(Itri3CalculateMR.class);
+
+                job.setMapperClass(HtMap.class);
+                job.setReducerClass(HtReduce.class);
+
+                job.setMapOutputKeyClass(Text.class);
+                job.setMapOutputValueClass(Text.class);
+
+                job.setInputFormatClass(TableInputFormat.class);
+                job.setOutputFormatClass(TableOutputFormat.class);
+
+                TableMapReduceUtil.initTableMapperJob(tablename, myScan, HtMap.class,
+                                Text.class, Text.class, job);
+                TableMapReduceUtil.initTableReducerJob(tablename, HtReduce.class, job);
+
+                System.exit(job.waitForCompletion(true) ? 0 : 1);
+
+        }
+}
+
+}}}