{{{
#!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);

	}
}

}}}