// Path: src/main/java/com/cloudera/knittingboar/io/InputRecordsSplit.java
// public class InputRecordsSplit {
//   
//   TextInputFormat input_format = null;
//   InputSplit split = null;
//   JobConf jobConf = null;
//   
//   RecordReader<LongWritable,Text> reader = null;
//   LongWritable key = null;
//   
//   final Reporter voidReporter = Reporter.NULL;
//   
//   public InputRecordsSplit(JobConf jobConf, InputSplit split)
//       throws IOException {
//     
//     this.jobConf = jobConf;
//     this.split = split;
//     this.input_format = new TextInputFormat();
//     
//     // RecordReader<LongWritable, Text> reader =
//     // format.getRecordReader(splits[x], job, reporter);
//     this.reader = input_format.getRecordReader(this.split, this.jobConf,
//         voidReporter);
//     this.key = reader.createKey();
//     // Text value = reader.createValue();
//     
//   }
//   
//   /**
//    * 
//    * just a dead simple way to do this
//    * 
//    * - functionality from TestTextInputFormat::readSplit()
//    * 
//    * If returns true, then csv_line contains the next line If returns false,
//    * then there is no next record
//    * 
//    * Will terminate when it hits the end of the split based on the information
//    * provided in the split class to the constructor and the TextInputFormat
//    * 
//    * @param csv_line
//    * @throws IOException
//    */
//   public boolean next(Text csv_line) throws IOException {
//     
//     return reader.next(key, csv_line);
//     
//   }
//   
//   public void ResetToStartOfSplit() throws IOException {
//     
//     // I'mma cheatin here. sue me.
//     this.reader = input_format.getRecordReader(this.split, this.jobConf,
//         voidReporter);
//     
//   }
//   
// }
// 
// Path: src/test/java/com/cloudera/knittingboar/utils/TestingUtils.java
// public class TestingUtils {
// 
//   public static void copyDecompressed(String resource, File output)
//       throws IOException {
//     URL input = Resources.getResource(resource);
//     ByteStreams.copy(new GZIPInputStream(input.openStream()),
//         new FileOutputStream(output));
//   }
// 
// }


import java.io.File;
import org.apache.commons.io.FileUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.mahout.classifier.sgd.L1;
import org.apache.mahout.classifier.sgd.OnlineLogisticRegression;
import org.apache.mahout.math.DenseVector;
import org.apache.mahout.math.RandomAccessSparseVector;
import org.apache.mahout.math.Vector;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import com.cloudera.knittingboar.io.InputRecordsSplit;
import com.cloudera.knittingboar.utils.TestingUtils;
import com.google.common.io.Files;
    // stepOffset, decay, and alpha --- describe how the learning rate decreases
    // lambda: amount of regularization
    // learningRate: amount of initial learning rate
    @SuppressWarnings("resource")
    OnlineLogisticRegression learningAlgorithm = new OnlineLogisticRegression(
        20, FEATURES, new L1()).alpha(1).stepOffset(1000).decayExponent(0.9)
        .lambda(3.0e-5).learningRate(20);

    FileInputFormat.setInputPaths(job, workDir);

    // try splitting the file in a variety of sizes
    TextInputFormat format = new TextInputFormat();
    format.configure(job);
    Text value = new Text();

    int numSplits = 1;

    InputSplit[] splits = format.getSplits(job, numSplits);

    LOG.info("requested " + numSplits + " splits, splitting: got =        "
        + splits.length);
    LOG.info("---- debug splits --------- ");
    rec_factory.Debug();
    int total_read = 0;

    for (int x = 0; x < splits.length; x++) {

      LOG.info("> Split [" + x + "]: " + splits[x].getLength());

      int count = 0;
// Path: src/main/java/com/cloudera/knittingboar/io/InputRecordsSplit.java
// public class InputRecordsSplit {
//   
//   TextInputFormat input_format = null;
//   InputSplit split = null;
//   JobConf jobConf = null;
//   
//   RecordReader<LongWritable,Text> reader = null;
//   LongWritable key = null;
//   
//   final Reporter voidReporter = Reporter.NULL;
//   
//   public InputRecordsSplit(JobConf jobConf, InputSplit split)
//       throws IOException {
//     
//     this.jobConf = jobConf;
//     this.split = split;
//     this.input_format = new TextInputFormat();
//     
//     // RecordReader<LongWritable, Text> reader =
//     // format.getRecordReader(splits[x], job, reporter);
//     this.reader = input_format.getRecordReader(this.split, this.jobConf,
//         voidReporter);
//     this.key = reader.createKey();
//     // Text value = reader.createValue();
//     
//   }
//   
//   /**
//    * 
//    * just a dead simple way to do this
//    * 
//    * - functionality from TestTextInputFormat::readSplit()
//    * 
//    * If returns true, then csv_line contains the next line If returns false,
//    * then there is no next record
//    * 
//    * Will terminate when it hits the end of the split based on the information
//    * provided in the split class to the constructor and the TextInputFormat
//    * 
//    * @param csv_line
//    * @throws IOException
//    */
//   public boolean next(Text csv_line) throws IOException {
//     
//     return reader.next(key, csv_line);
//     
//   }
//   
//   public void ResetToStartOfSplit() throws IOException {
//     
//     // I'mma cheatin here. sue me.
//     this.reader = input_format.getRecordReader(this.split, this.jobConf,
//         voidReporter);
//     
//   }
//   
// }
// 
// Path: src/test/java/com/cloudera/knittingboar/utils/TestingUtils.java
// public class TestingUtils {
// 
//   public static void copyDecompressed(String resource, File output)
//       throws IOException {
//     URL input = Resources.getResource(resource);
//     ByteStreams.copy(new GZIPInputStream(input.openStream()),
//         new FileOutputStream(output));
//   }
// 
// }


// Path: src/test/java/com/cloudera/knittingboar/records/TestTwentyNewsgroupsCustomRecordParseOLRRun.java
import java.io.File;
import org.apache.commons.io.FileUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.mahout.classifier.sgd.L1;
import org.apache.mahout.classifier.sgd.OnlineLogisticRegression;
import org.apache.mahout.math.DenseVector;
import org.apache.mahout.math.RandomAccessSparseVector;
import org.apache.mahout.math.Vector;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import com.cloudera.knittingboar.io.InputRecordsSplit;
import com.cloudera.knittingboar.utils.TestingUtils;
import com.google.common.io.Files;

    // stepOffset, decay, and alpha --- describe how the learning rate decreases
    // lambda: amount of regularization
    // learningRate: amount of initial learning rate
    @SuppressWarnings("resource")
    OnlineLogisticRegression learningAlgorithm = new OnlineLogisticRegression(
        20, FEATURES, new L1()).alpha(1).stepOffset(1000).decayExponent(0.9)
        .lambda(3.0e-5).learningRate(20);

    FileInputFormat.setInputPaths(job, workDir);

    // try splitting the file in a variety of sizes
    TextInputFormat format = new TextInputFormat();
    format.configure(job);
    Text value = new Text();

    int numSplits = 1;

    InputSplit[] splits = format.getSplits(job, numSplits);

    LOG.info("requested " + numSplits + " splits, splitting: got =        "
        + splits.length);
    LOG.info("---- debug splits --------- ");
    rec_factory.Debug();
    int total_read = 0;

    for (int x = 0; x < splits.length; x++) {

      LOG.info("> Split [" + x + "]: " + splits[x].getLength());

      int count = 0;
      InputRecordsSplit custom_reader = new InputRecordsSplit(job, splits[x]);