Index: /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/PagedJSON.java
===================================================================
--- /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/PagedJSON.java	(revision 30936)
+++ /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/PagedJSON.java	(revision 30937)
@@ -1,20 +1,8 @@
 package org.hathitrust;
 
-import java.io.BufferedInputStream;
 import java.io.BufferedReader;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.net.URI;
 import java.util.ArrayList;
 import java.util.Iterator;
 
-import org.apache.commons.compress.compressors.CompressorException;
-import org.apache.commons.compress.compressors.CompressorInputStream;
-import org.apache.commons.compress.compressors.CompressorStreamFactory;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FSDataInputStream;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
 import org.apache.spark.api.java.function.FlatMapFunction;
 import org.json.JSONArray;
@@ -42,40 +30,4 @@
 	}
 	
-	protected static BufferedInputStream getBufferedInputStream(String fileIn) 
-			throws IOException 
-	{
-	    BufferedInputStream bis = null;
-	    
-	    if (fileIn.startsWith("hdfs://")) {
-	    	URI uri = URI.create (fileIn);
-	    	Configuration conf = new Configuration();
-	    	FileSystem file = FileSystem.get(uri, conf);
-	    	FSDataInputStream fin = file.open(new Path(uri));
-
-	    	bis = new BufferedInputStream(fin);
-	    }
-	    else {
-	    	// Trim 'file://' off the front
-	    	
-	    	String local_file_in = fileIn;
-	    	if (local_file_in.startsWith("file://")) {
-	    		local_file_in = fileIn.substring("file://".length());
-	    	}
-	    	FileInputStream fin = new FileInputStream(local_file_in);
-	    	bis = new BufferedInputStream(fin);
-	    }
-
-	    return bis;
-		
-	}
-	protected static BufferedReader getBufferedReaderForCompressedFile(String fileIn) 
-			throws IOException, CompressorException 
-	{
-	    BufferedInputStream bis = getBufferedInputStream(fileIn);
-	    CompressorInputStream comp_input = new CompressorStreamFactory().createCompressorInputStream(bis);
-	    BufferedReader br = new BufferedReader(new InputStreamReader(comp_input,"UTF8"));
-	    return br;
-	}
-	
 	protected JSONObject readJSONFile(String filename)
 	{
@@ -87,5 +39,5 @@
 
 			String str;
-			BufferedReader br = getBufferedReaderForCompressedFile(_input_dir + "/" + filename);
+			BufferedReader br = ClusterFileIO.getBufferedReaderForCompressedFile(_input_dir + "/" + filename);
 			while ((str = br.readLine()) != null) {
 				sb.append(str);
@@ -120,7 +72,10 @@
 	}
 
-	public Iterator<String> call(String s) 
+	public Iterator<String> call(String json_file_in) 
 	{ 
-		JSONObject extracted_feature_record = readJSONFile(s);
+		JSONObject extracted_feature_record = readJSONFile(json_file_in);
+		
+		// Check output directory for volume exists, and create it if not
+		
 		
 		String id = extracted_feature_record.getString("id");
Index: /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/PrepareForIngest.java
===================================================================
--- /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/PrepareForIngest.java	(revision 30936)
+++ /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/PrepareForIngest.java	(revision 30937)
@@ -19,9 +19,10 @@
 	private static final long serialVersionUID = 1L;
 
-    protected String _input_dir;
-    protected String _json_list_filename;
-    protected String _output_dir;
-    protected int    _verbosity;
-    
+	protected String _input_dir;
+	protected String _json_list_filename;
+	protected String _output_dir;
+	
+	protected int    _verbosity;
+
 	public PrepareForIngest(String input_dir, String json_list_filename, String output_dir, int verbosity)
 	{
@@ -34,16 +35,23 @@
 	public void exec()
 	{	
-		SparkConf conf = new SparkConf().setAppName("HTRC-EF: Prepare for Solr Ingest");
-		JavaSparkContext sc = new JavaSparkContext(conf);
+		String spark_app_name = "HathiTrust Extract Features: Prepare for Solr Ingest";
+		spark_app_name += "[" + _json_list_filename + "]";
 
-		JavaRDD<String> json_list_data = sc.textFile(_json_list_filename).cache();
+		SparkConf conf = new SparkConf().setAppName(spark_app_name);
+		JavaSparkContext jsc = new JavaSparkContext(conf);
+		ClusterFileIO.init(_input_dir);
+		
+		// Check output directory exists, and create it if not
+	
+				
+		JavaRDD<String> json_list_data = jsc.textFile(_json_list_filename).cache();
 
 		JavaRDD<String> json_ids = json_list_data.flatMap(new PagedJSON(_input_dir));
-		
+
 
 		//long numAs = json_list_data.filter(new ContainsA()).count();
 
-	
-/*
+
+		/*
 		long numBs = json_list_data.filter(new Function<String, Boolean>() {
 			public Boolean call(String s) { return s.contains("b"); }
@@ -51,5 +59,5 @@
 
 		System.out.println("#### Lines with a: " + numAs + ", lines with b: " + numBs);
-	*/
+		 */
 		long num_ids = json_ids.count();
 		System.out.println("");
@@ -58,6 +66,6 @@
 		System.out.println("############");
 		System.out.println("");
-		
-		sc.close();
+
+		jsc.close();
 	}
 
@@ -90,9 +98,5 @@
 		catch (ParseException e) {
 			System.err.println(e.getMessage());
-			//System.err.println("Usage: RUN.bat [options] json-file-list.txt input-dir output-dir");
-			formatter.printHelp("RUN.bash/RUN.bat [options] json-file-list.txt input-dir output-dir", options);
-			//System.err.println("  Where 'filename.txt' contains a list of JSON files, one per line,");
-			//System.err.println("  which use the HathiTrust Extracted Feature JSON format");
-
+			formatter.printHelp("RUN.bash [options] json-file-list.txt input-dir output-dir", options);
 			System.exit(1);
 			return;
@@ -110,20 +114,14 @@
 
 
+		String[] filtered_args = cmd.getArgs();
 
-		String[] filtered_args = cmd.getArgs();
-		
 		if (filtered_args.length != 3) {
-		        //System.err.println("Usage: RUN.bat [options] json-filelist.txt input-dir output-dir");
-		        formatter.printHelp("RUN.bash/RUN.bat  [options] json-filelist.txt input-dir output-dir", options);
-
-			//System.err.println("Usage: RUN.bat [options] input-dir output-dir");
-			//System.err.println("  Where 'filename.txt' contains a list of JSON files, one per line,");
-			//System.err.println("  which use the HathiTrust Extracted Feature JSON format");
-		        System.exit(1);
+			formatter.printHelp("RUN.bash [options] json-filelist.txt input-dir output-dir", options);
+			System.exit(1);
 		}
 		String json_list_filename = filtered_args[0];
 		String input_dir  = filtered_args[1];
 		String output_dir = filtered_args[2];
-		
+
 
 		//String json_list_filename = cmd.getArgs()[0]; // args[0];
