{{{
#!java
package tsmc;

// TSMC serial program number 4
// 0. after TSMC3CalculateMR
// 1. run addIndexToTurnover() would add index table to tsmc.
//    if you have indexed the table before, don't do it again.
// 2. tune the filter_value in main() and run readSortedValGreater() would show report

import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.tableindexed.IndexSpecification;
import org.apache.hadoop.hbase.client.tableindexed.IndexedTable;
import org.apache.hadoop.hbase.client.tableindexed.IndexedTableAdmin;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

public class TSMC4SortTurnover {

	public void addIndexToTurnover(String OriTable, String IndexID,
			String OriColumn) throws IOException {
		HBaseConfiguration conf = new HBaseConfiguration();
		conf.addResource(new Path("/opt/hbase/conf/hbase-site.xml"));
		IndexedTableAdmin admin = new IndexedTableAdmin(conf);
		admin.addIndex(Bytes.toBytes(OriTable), new IndexSpecification(IndexID,
				Bytes.toBytes(OriColumn)));

	}

	public void readSortedValGreater(String filter_val) throws IOException {
		HBaseConfiguration conf = new HBaseConfiguration();
		conf.addResource(new Path("/opt/hbase/conf/hbase-site.xml"));
		// the id of the index to use
		String tablename = "tsmc";
		String indexId = "Sum";
		byte[] column_1 = Bytes.toBytes("Turnover:Sum");
		byte[] column_2 = Bytes.toBytes("Detail:Name");

		byte[] indexStartRow = HConstants.EMPTY_START_ROW;
		byte[] indexStopRow = null;
		byte[][] indexColumns = null;

		SingleColumnValueFilter indexFilter = new SingleColumnValueFilter(Bytes
				.toBytes("Turnover"), Bytes.toBytes("Sum"),
				CompareFilter.CompareOp.GREATER_OR_EQUAL, Bytes
						.toBytes(filter_val));

		byte[][] baseColumns = new byte[][] { column_1, column_2 };

		IndexedTable table = new IndexedTable(conf, Bytes.toBytes(tablename));

		ResultScanner scanner = table.getIndexedScanner(indexId, indexStartRow,
				indexStopRow, indexColumns, indexFilter, baseColumns);

		for (Result rowResult : scanner) {
			String sum = Bytes.toString(rowResult.getValue(column_1));
			String name = Bytes.toString(rowResult.getValue(column_2));
			System.out.println(name + " 's turnover is " + sum + " $.");
		}
		table.close();
	}

	public static void main(String[] args) throws IOException {

		TSMC4SortTurnover tt = new TSMC4SortTurnover();
		tt.addIndexToTurnover("tsmc", "Sum", "Turnover:Sum");
		tt.readSortedValGreater("130");

	}
}

}}}