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

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[] values = line.split("\t", -1);
                    String title = values[0];
                    String productId = values[1];
                    String userId = values[2];
                    String text = values[3];
                    long timestamp = Long.parseLong(values[4]);
                    float score = Float.parseFloat(values[5]);

                    JSONObject json = new JSONObject();
                    json.put("productId", productId);
                    json.put("userId", userId);
                    json.put("text", text);
                    json.put("timestamp", timestamp);
                    json.put("score", score);

                    if (title.length() > 0) {
                        byte[] rowKey = Bytes.toBytes(title);
                        byte[] colKey = Bytes.toBytes(timestamp + "_" + userId);
                        byte[] colValue = Bytes.toBytes(json.toJSONString());

                        Put put = new Put(rowKey);
                        put.add(Bytes.toBytes("cf1"), colKey, colValue);
                        htable.put(put);
                    }

                    line = reader.readLine();

                    if (i % 10000 == 0) {
                        System.out.println("Wrote record " + i);
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(1);
        } finally {
            if (htable != null) htable.close();
            if (reader != null) reader.close();
        }
    }
}

@@ -459,22 +459,90 @@
 ple.JSON
-Object
+Array;%0Aimport org.json.simple.JSONObject;%0Aimport org.json.simple.JSONValue
 ;%0A%0Apubli
@@ -1564,22 +1564,19 @@
 tring%5B%5D 
-values
+row
  = line.
@@ -1616,32 +1616,45 @@
     
-String title = values
+byte%5B%5D rowKey = Bytes.toBytes(row
 %5B0%5D
+)
 ;%0A  
@@ -1675,36 +1675,43 @@
     
-String productId = values
+Object obj =JSONValue.parse(row
 %5B1%5D
+)
 ;%0A  
@@ -1732,144 +1732,94 @@
     
-String userId = values%5B2%5D;%0A                    String text = values%5B3%5D;%0A                    long timestamp = Long.parseLong(values%5B4%5D
+JSONArray jsonList = (JSONArray)obj;%0A%0A                    Put put = new Put(rowKey
 );%0A
+%0A
     
@@ -1839,51 +1839,55 @@
    f
-loat score = Float.parseFloat(values%5B5%5D);%0A%0A
+or (int j = 0; j %3C jsonList.size(); j++) %7B%0A    
     
@@ -1917,23 +1917,23 @@
 ect json
- = new 
+Obj = (
 JSONObje
@@ -1938,66 +1938,27 @@
 ject
-();%0A                    json.put(%22productId%22, productId
+)jsonList.get(j
 );%0A
+%0A
     
@@ -1977,107 +1977,45 @@
     
-json.put(%22userId%22, userId);%0A                    json.put(%22text%22, text);%0A                    json.pu
+    long timestamp = (Long)jsonObj.ge
 t(%22t
@@ -2027,19 +2027,8 @@
 amp%22
-, timestamp
 );%0A 
@@ -2050,140 +2050,56 @@
     
-json.put(%22score%22, score);%0A%0A                    if (title.length() %3E 0) %7B%0A                        byte%5B%5D rowKey = Bytes.toBytes(title
+    String userId = (String)jsonObj.get(%22userId%22
 );%0A 
@@ -2238,16 +2238,19 @@
 tes(json
+Obj
 .toJSONS
@@ -2264,59 +2264,8 @@
 );%0A%0A
-                        Put put = new Put(rowKey);%0A
     
@@ -2357,28 +2357,10 @@
     
-    htable.put(put);
+%7D%0A
 %0A   
@@ -2368,33 +2368,48 @@
                 
-%7D
+htable.put(put);
 %0A%0A              
@@ -2443,138 +2443,8 @@
 e();
-%0A%0A                    if (i %25 10000 == 0) %7B%0A                        System.out.println(%22Wrote record %22 + i);%0A                    %7D
 %0A   

Improve bulk loading of hbase by using new results of map reduce to add all columns for a row in one put call.