<commit_before>package tv.floe.canova.exec.serial;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.Properties;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;



public class SerialExecDriver {
	
	final Reporter voidReporter = Reporter.NULL;
	protected static JobConf defaultConf = new JobConf();
	private static FileSystem localFs = null;
	static {
		try {
			defaultConf.set("fs.defaultFS", "file:///");
			localFs = FileSystem.getLocal(defaultConf);
		} catch (IOException e) {
			throw new RuntimeException("init failure", e);
		}
	}

	private static Path workDir = new Path("/tmp/");

	Properties props;

	private String app_properties_file = "";

	protected InputSplit[] splits;	
	
	InputFormat inputFormat = null;
	
	/**
	 * TODO: Instantiate the input format from the property file's conf'd class
	 * 
	 * 
	 * @param input_path
	 * @param job
	 * @return
	 */
	private InputSplit[] computeSplits(Path input_path, JobConf job) {

		long block_size = localFs.getDefaultBlockSize();

		System.out.println("default block size: " + (block_size / 1024 / 1024)
				+ "MB");

		// ---- set where we'll read the input files from -------------
		FileInputFormat.setInputPaths(job, input_path);

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

		int numSplits = 1;

		InputSplit[] splits = null;

		try {
			splits = format.getSplits(job, numSplits);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		return splits;

	}	

	public RecordReader getRecordReaderForSplit( InputSplit split, JobConf job ) {
		
		try {
			return this.inputFormat.getRecordReader( split, job, voidReporter);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		return null;
		
	}

	/**
	 * Setup components of the IR app run 1. load app.properties 2. msg arrays
	 * 3. calc local splits 4. setup master 5. setup workers based on number of
	 * splits
	 * 
	 */
	public void setup() {

		// ----- load the app.properties file

		// String configFile = (args.length < 1) ?
		// ConfigFields.DEFAULT_CONFIG_FILE : args[0];
		this.props = new Properties();
		// Configuration conf = getConf();

		try {
			FileInputStream fis = new FileInputStream(this.app_properties_file);
			props.load(fis);
			fis.close();
		} catch (FileNotFoundException ex) {
			// throw ex; // TODO: be nice
			System.out.println(ex);
		} catch (IOException ex) {
			// throw ex; // TODO: be nice
			System.out.println(ex);
		}

		// setup msg arrays

		// calc splits

		// ---- this all needs to be done in
		JobConf job = new JobConf(defaultConf);
		
		// app.input.path
		
		Path splitPath = new Path( props.getProperty("app.input.path") );

		System.out.println( "app.input.path = " + splitPath );
		
		// TODO: work on this, splits are generating for everything in dir
		InputSplit[] splits = computeSplits(splitPath, job);

		System.out.println("split count: " + splits.length);



	}	
	
	/**
	 * TODO:
	 * - read the props file, setup
	 * - compute the inputs
	 * - for each input -> open a file reader
	 * - connect the file reader to an InputFormat based on the conf/props
	 * - Create an instance of the MapTask
	 * - read K/V pairs from the input format's RecordReader til EOF
	 * 
	 */
	public static void run() {
		
		SerialExecDriver driver = new SerialExecDriver();
		
		// read props
		// compute splist
		driver.setup();
		
		// setup MapTask->Vectorizer engine
		
		for ( int x = 0; x < driver.splits.length; x++ ) {
			
			// get a RecordReader for split
			RecordReader rr = driver.getRecordReaderForSplit(driver.splits[x], driver.defaultConf );
			// for each record, 
			
			while (rr.getProgress() < 1.0) {
				
				// get the next k/v pair
				rr.n
				
				// feed it to the map task harness w the vector engine
				
			}
			
		}
		
		
		
	}

}
<commit_msg>rebuilt serial driver, still a WIP of skeleton code<commit_after>package tv.floe.canova.exec.serial;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.Properties;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;

import tv.floe.canova.vector.engine.VectorMapTaskEngine;



public class SerialExecDriver {
	
	final Reporter voidReporter = Reporter.NULL;
	protected static JobConf defaultConf = new JobConf();
	private static FileSystem localFs = null;
	static {
		try {
			defaultConf.set("fs.defaultFS", "file:///");
			localFs = FileSystem.getLocal(defaultConf);
		} catch (IOException e) {
			throw new RuntimeException("init failure", e);
		}
	}

	private static Path workDir = new Path("/tmp/");

	Properties props;

	private String app_properties_file = "";

	protected InputSplit[] splits;	
	
	String inputFormatClassname = "org.apache.hadoop.mapred.TextInputFormat"; //"tv.floe.canova.input.format";
	InputFormat inputFormat = null;
	
	public SerialExecDriver( String propsFile ) {
		
		this.app_properties_file = propsFile;
		
	}
	
	/**
	 * TODO: Instantiate the input format from the property file's conf'd class
	 * 
	 * 
	 * @param input_path
	 * @param job
	 * @return
	 */
	private InputSplit[] computeSplits(Path input_path, JobConf job) {

		long block_size = localFs.getDefaultBlockSize();

		System.out.println("default block size: " + (block_size / 1024 / 1024)
				+ "MB");

		// ---- set where we'll read the input files from -------------
		FileInputFormat.setInputPaths(job, input_path);

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

		int numSplits = 1;

		InputSplit[] splits = null;

		try {
			splits = format.getSplits(job, numSplits);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		return splits;

	}	

	public RecordReader getRecordReaderForSplit( InputSplit split, JobConf job ) {
		
		try {
			return this.inputFormat.getRecordReader( split, job, voidReporter);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		return null;
		
	}

	/**
	 * Setup components of the IR app run 1. load app.properties 2. msg arrays
	 * 3. calc local splits 4. setup master 5. setup workers based on number of
	 * splits
	 * 
	 */
	public void setup() {

		// ----- load the app.properties file

		// String configFile = (args.length < 1) ?
		// ConfigFields.DEFAULT_CONFIG_FILE : args[0];
		this.props = new Properties();
		// Configuration conf = getConf();

		try {
			FileInputStream fis = new FileInputStream(this.app_properties_file);
			props.load(fis);
			fis.close();
		} catch (FileNotFoundException ex) {
			// throw ex; // TODO: be nice
			System.out.println(ex);
		} catch (IOException ex) {
			// throw ex; // TODO: be nice
			System.out.println(ex);
		}
		
		// get input format class
		
		Class<?> inputFormat_clazz;
		try {
			inputFormat_clazz = Class.forName(props
					.getProperty("tv.floe.canova.input.format"));
			
			Constructor<?> inputFormat_ctor = inputFormat_clazz.getConstructor();
			
			this.inputFormat = (InputFormat) inputFormat_ctor.newInstance(); // new
																	// Object[]
																	// {
																	// ctorArgument
																	// });

		} catch (ClassNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (SecurityException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (NoSuchMethodException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IllegalArgumentException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (InstantiationException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IllegalAccessException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (InvocationTargetException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		
		

		// setup msg arrays

		// calc splits

		// ---- this all needs to be done in
		JobConf job = new JobConf(defaultConf);
		
		// app.input.path
		
		
		
		Path splitPath = new Path( props.getProperty("tv.floe.canova.input.path") );

		System.out.println( "app.input.path = " + splitPath );
		
		// TODO: work on this, splits are generating for everything in dir
		this.splits = computeSplits(splitPath, job);

		System.out.println("split count: " + splits.length);



	}	
	
	/**
	 * TODO:
	 * - read the props file, setup
	 * - compute the inputs
	 * - for each input -> open a file reader
	 * - connect the file reader to an InputFormat based on the conf/props
	 * - Create an instance of the MapTask
	 * - read K/V pairs from the input format's RecordReader til EOF
	 * 
	 * 
	 * 
	 * TODO: questions
	 * - how do we infer or setup the template variables?
	 * @throws IOException 
	 * 
	 */
	public static void run() throws IOException {
		
		SerialExecDriver driver = new SerialExecDriver( "src/test/resources/single_process.unit_test.properties" );
		VectorMapTaskEngine vector_engine = new VectorMapTaskEngine();
		
		// read props
		// compute splist
		driver.setup();
		
		// setup MapTask->Vectorizer engine
		
		for ( int x = 0; x < driver.splits.length; x++ ) {
			
			// get a RecordReader for split
			RecordReader rr = driver.getRecordReaderForSplit(driver.splits[x], driver.defaultConf );
			// for each record, 
			
			LongWritable key = (LongWritable) rr.createKey();
			Text val = (Text) rr.createValue();
			
			while (rr.getProgress() < 1.0) {
				
				// get the next k/v pair
				boolean gotKey = rr.next( key, val );
				
				// feed it to the map task harness w the vector engine
				
				System.out.println( "line: " + val.toString() ); 
				
			}
			
		}
		
		
		
	}
	
	public static void main(String[] args) throws IOException {

		SerialExecDriver.run();
		
	}

}

{'fraction_non_alphanumeric': 0.11608122941822173, 'fraction_numerical': 0.0034760336626417855, 'mean_word_length': 3.0477600888559793, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 9, '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': '416652', 'n_tokens_mistral': 3902, 'n_tokens_neox': 3214, 'n_words': 931}