package com.isuraed.insight;

import java.io.BufferedReader;
import java.io.InputStreamReader;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.util.Bytes;

import org.json.simple.JSONArray;
import org.json.simple.JSONObject;
import org.json.simple.JSONValue;

public class InsertReviews {

    // Insert records into HBase reviews table from all files in the specified HDFS folder. Input path is the full
    // hdfs path including hostname - i.e. hdfs://hostname:9000/path/to/files.
    public static void main(String[] args) throws Exception {
        if (args.length != 2) {
            System.out.println("Usage: com.isuraed.insight.InsertReviews <hdfs_input_path> <hbase_table_name>");
            System.exit(1);
        }

        HTable htable = null;
        BufferedReader reader = null;
        try {
            FileSystem fs = FileSystem.get(new Configuration());
            FileStatus[] status = fs.listStatus(new Path(args[0]));

            htable = new HTable(HBaseConfiguration.create(), args[1]);

            for (int i = 0; i < status.length; i++) {
                reader = new BufferedReader(new InputStreamReader(fs.open(status[i].getPath())));

                String line = reader.readLine();
                while (line != null) {
                    String[] row = line.split("\t", -1);
                    byte[] rowKey = Bytes.toBytes(row[0]);
                    Object obj =JSONValue.parse(row[1]);
                    JSONArray jsonList = (JSONArray)obj;

                    Put put = new Put(rowKey);

                    for (int j = 0; j < jsonList.size(); j++) {
                        JSONObject jsonObj = (JSONObject)jsonList.get(j);

                        long timestamp = (Long)jsonObj.get("timestamp");
                        String userId = (String)jsonObj.get("userId");
                        byte[] colKey = Bytes.toBytes(timestamp + "_" + userId);
                        byte[] colValue = Bytes.toBytes(jsonObj.toJSONString());

                        put.add(Bytes.toBytes("cf1"), colKey, colValue);
                    }

                    htable.put(put);

                    line = reader.readLine();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(1);
        } finally {
            if (htable != null) htable.close();
            if (reader != null) reader.close();
        }
    }
}

@@ -87,16 +87,49 @@
 mReader;
+%0Aimport java.util.logging.Logger;
 %0A%0Aimport
@@ -281,16 +281,66 @@
 s.Path;%0A
+import org.apache.hadoop.hbase.client.HBaseAdmin;%0A
 import o
@@ -463,32 +463,131 @@
 eConfiguration;%0A
+import org.apache.hadoop.hbase.HColumnDescriptor;%0Aimport org.apache.hadoop.hbase.HTableDescriptor;%0A
 import org.apach
@@ -1200,285 +1200,964 @@
     
-HTable htable = null;%0A        BufferedReader reader = null;%0A        try %7B%0A            FileSystem fs = FileSystem.get(new Configuration());%0A            FileStatus%5B%5D status = fs.listStatus(new Path(args%5B0%5D));%0A%0A            htable = new HTable(HBaseConfiguration.create(), args%5B1%5D)
+Logger logger = Logger.getLogger(InsertReviews.class.getName());%0A%0A        Configuration conf = HBaseConfiguration.create();%0A        HBaseAdmin hadmin = new HBaseAdmin(conf);%0A        HTableDescriptor hdesc = new HTableDescriptor(args%5B1%5D);%0A        HColumnDescriptor hfamily = new HColumnDescriptor(%22cf1%22);%0A        hdesc.addFamily(hfamily);%0A%0A        HTable htable = null;%0A        BufferedReader reader = null;%0A        try %7B%0A            FileSystem fs = FileSystem.get(new Configuration());%0A            FileStatus%5B%5D status = fs.listStatus(new Path(args%5B0%5D));%0A%0A            if (hadmin.tableExists(args%5B1%5D)) %7B%0A                hadmin.disableTable(args%5B1%5D);%0A                hadmin.deleteTable(args%5B1%5D);%0A                logger.info(%22Deleted existing table %22 + args%5B1%5D);%0A            %7D%0A%0A            hadmin.createTable(hdesc);%0A            logger.info(%22Created table %22 + args%5B1%5D);%0A%0A            htable = new HTable(conf, hdesc.getTableName());%0A            int rowCount = 0
 ;%0A%0A 
@@ -2306,16 +2306,92 @@
 th())));
+%0A                logger.info(%22Processing HDFS file %22 + status%5Bi%5D.getPath());
 %0A%0A      
@@ -3233,27 +3233,24 @@
 add(
-Bytes.toBytes(%22cf1%22
+hfamily.getName(
 ), c
@@ -3326,16 +3326,185 @@
 ut(put);
+%0A                    rowCount++;%0A%0A                    if (rowCount %25 10000 == 0) %7B%0A                        logger.info(rowCount + %22 rows written%22);%0A                    %7D
 %0A%0A      

Delete and re-create HBase table before insert.