package itri;



// ITRI serial program number 5

// 0. after Itri4SortTurnover

// run it



import java.io.IOException;



import org.apache.hadoop.conf.Configuration;

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.IndexedTable;

import org.apache.hadoop.hbase.filter.CompareFilter;

import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;

import org.apache.hadoop.hbase.util.Bytes;

import org.apache.hadoop.util.GenericOptionsParser;



public class Itri5ShowReport {



  static 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 = "itri";

    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[] argv) throws IOException {



    String[] args = new GenericOptionsParser(new Configuration(), argv)

        .getRemainingArgs();

    if (args.length < 1) {

      System.out.println("Input the MIN number:");

      return;

    }

    readSortedValGreater(args[0]);

  }

}