package com.isuraed.insight;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
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 {

    public static void main(String[] args) throws Exception {
        if (args.length != 1) {
            System.out.println("Usage: com.isuraed.insight.InsertReviews <hdfs_input_path>");
            System.exit(1);
        }

        String hdfsFile = args[0];
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(hdfsFile), conf);
        Path hdfsPath = new Path(hdfsFile);

        if (!fs.exists(hdfsPath)) {
            System.out.println("File " + hdfsFile + " does not exist.");
            System.exit(1);
        }

        Configuration hconf = HBaseConfiguration.create();
        HTable htable = new HTable(hconf, "isura_reviews");
        BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(hdfsPath)));

        String line = br.readLine();
        while (line != null) {
            String[] values = line.split("\t");
            String title = values[0];
            String productId = values[1];
            String userId = values[2];
            long timestamp = Long.parseLong(values[3]);
            float score = Float.parseFloat(values[4]);
            float averageScore = Float.parseFloat(values[5]);
            String text = values[6];

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

            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 = br.readLine();
        }

        br.close();
        htable.close();
    }
}

@@ -1606,40 +1606,22 @@
     
-long timestamp = Long.parseLong(
+String text = 
 valu
@@ -1625,17 +1625,16 @@
 alues%5B3%5D
-)
 ;%0A      
@@ -1643,38 +1643,39 @@
     
-f
 lo
-at score = Float
+ng timestamp = Long
 .parse
-Float
+Long
 (val
@@ -1701,24 +1701,17 @@
   float 
-averageS
+s
 core = F
@@ -1741,45 +1741,8 @@
 5%5D);
-%0A            String text = values%5B6%5D;
 %0A%0A  
@@ -1865,32 +1865,68 @@
 erId%22, userId);%0A
+            json.put(%22text%22, text);%0A
             json
@@ -1955,16 +1955,16 @@
 stamp);%0A
+
         
@@ -1996,91 +1996,8 @@
 re);
-%0A            json.put(%22average%22, averageScore);%0A            json.put(%22text%22, text);
 %0A%0A  

Insert reviews into HBase.