<commit_before>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();
        }
    }
}<commit_msg>Improve bulk loading of hbase by using new results of map reduce to add all columns for a row in one put call.<commit_after>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();
        }
    }
}
{'fraction_non_alphanumeric': 0.11832513985421257, 'fraction_numerical': 0.006780810306831667, 'mean_word_length': 1.8015194681861348, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5217785', 'n_tokens_mistral': 1652, 'n_tokens_neox': 1571, 'n_words': 372}