Index: other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/ClusterFileIO.java
===================================================================
--- other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/ClusterFileIO.java	(revision 30944)
+++ other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/ClusterFileIO.java	(revision 30945)
@@ -21,5 +21,4 @@
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
-import org.apache.spark.api.java.JavaSparkContext;
 
 public class ClusterFileIO {
@@ -58,5 +57,5 @@
 	*/
 	
-	public static FileSystem getFileSystemInstance(String input_dir)
+	protected static FileSystem getFileSystemInstance(String input_dir)
 	{
 		if (__fs == null) {
@@ -82,5 +81,5 @@
 	}
 	
-	protected static boolean exists(String file) 
+	public static boolean exists(String file) 
 	{
 		FileSystem fs = getFileSystemInstance(file);
@@ -100,5 +99,32 @@
 	    return exists;
 	}
-	protected static BufferedInputStream getBufferedInputStream(String fileIn) 
+	
+	public static String removeSuffix(String file,String suffix)
+	{
+		return file.substring(0,file.length() - suffix.length());
+	}
+	
+	public static boolean createDirectoryAll(String dir) 
+	{
+		FileSystem fs = getFileSystemInstance(dir);
+		boolean created_dir = false;
+
+		if (!exists(dir)) {
+			try {
+				URI uri = new URI(dir);
+				Path path = new Path(uri);
+				fs.mkdirs(path);
+				created_dir = true;	
+			} catch (URISyntaxException e) {
+				e.printStackTrace();
+			} catch (IOException e) {
+				e.printStackTrace();
+			}
+		}
+		
+		return created_dir;
+	}
+	
+	public static BufferedInputStream getBufferedInputStream(String fileIn) 
 			throws IOException 
 	{
@@ -137,5 +163,5 @@
 	}
 
-	protected static BufferedOutputStream getBufferedOutputStream(String fileOut) 
+	public static BufferedOutputStream getBufferedOutputStream(String fileOut) 
 			throws IOException 
 	{
@@ -163,5 +189,5 @@
 	}
 	
-	protected static BufferedReader getBufferedReaderForCompressedFile(String fileIn) 
+	public static BufferedReader getBufferedReaderForCompressedFile(String fileIn) 
 			throws IOException, CompressorException 
 	{
@@ -172,5 +198,5 @@
 	}
 
-	protected static BufferedWriter getBufferedWriterForCompressedFile(String fileOut) 
+	public static BufferedWriter getBufferedWriterForCompressedFile(String fileOut) 
 			throws IOException, CompressorException 
 	{
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 30944)
+++ other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/PagedJSON.java	(revision 30945)
@@ -24,10 +24,12 @@
 
 	protected String _input_dir;
+	protected String _output_dir;
 	protected int    _verbosity;
 	
-	public PagedJSON(String input_dir, int verbosity)
+	public PagedJSON(String input_dir, String output_dir, int verbosity)
 	{
-		_input_dir = input_dir;
-		_verbosity = verbosity;
+		_input_dir  = input_dir;
+		_output_dir = output_dir;
+		_verbosity  = verbosity;
 	}
 	
@@ -44,21 +46,7 @@
 			while ((str = br.readLine()) != null) {
 				sb.append(str);
-			    //System.out.println(str);
 			}
 
 	        br.close();
-	        
-	        //System.err.println("*****" + sb.toString());
-	               
-	        /*
-			List<String> lines = Files.readAllLines(path,StandardCharsets.UTF_8);
-
-
-			for (String line : lines) {
-				sb.append(line);
-
-			}
-			*/
-	        
 		} 
 		catch (Exception e) {
@@ -68,8 +56,5 @@
 		JSONObject json_obj = new JSONObject(sb.toString());
 		
-		
 		return json_obj;
-		
-		//return sb.toString();
 	}
 
@@ -83,5 +68,5 @@
 		String id = extracted_feature_record.getString("id");
 		
-		JSONObject ef_metadata = extracted_feature_record.getJSONObject("metadata");
+		//JSONObject ef_metadata = extracted_feature_record.getJSONObject("metadata");
 		JSONObject ef_features = extracted_feature_record.getJSONObject("features");
 		
@@ -97,7 +82,27 @@
 		int ef_num_pages = ef_pages.length();
 		
+		// Make directory for page-level JSON output
+		String json_dir = ClusterFileIO.removeSuffix(json_file_in,".json.bz2");
+		String page_json_dir = json_dir + "/pages";
+		//ClusterFileIO.createDirectoryAll(_output_dir + "/" + page_json_dir);
+		System.out.println("mkdir: " + _output_dir + "/" + page_json_dir);
+		
 		ArrayList<String> ids = new ArrayList<String>(ef_num_pages);
 		for (int i = 0; i < ef_page_count; i++) {
-			ids.add(id + "." + i);
+			String formatted_i = String.format("page-%06d", i);
+			String page_id = id + "." + formatted_i;
+		
+			if (_verbosity >= 2) {
+			  System.out.println("  Page: " + page_id);
+			}
+			
+			// create JSON obj of just the page (for now)
+			// write it out
+			
+			ids.add(page_json_dir +"/" + page_id + ".json.bz2");
+			
+			if (i==0) {
+				System.out.println("Sample output JSON page file: " + page_json_dir +"/" + page_id + ".json.bz2");
+			}
 		}
 		
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 30944)
+++ other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/PrepareForIngest.java	(revision 30945)
@@ -1,17 +1,9 @@
 package org.hathitrust;
 
-import java.io.IOException;
 import java.io.Serializable;
-import java.nio.charset.StandardCharsets;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.util.List;
-
 import org.apache.commons.cli.*;
 
 import org.apache.spark.api.java.*;
 import org.apache.spark.SparkConf;
-import org.apache.spark.api.java.function.Function;
 
 public class PrepareForIngest implements Serializable
@@ -19,4 +11,6 @@
 	private static final long serialVersionUID = 1L;
 
+	public static final int NUM_PARTITIONS = 6; // default would appear to be 2
+	
 	protected String _input_dir;
 	protected String _json_list_filename;
@@ -40,29 +34,17 @@
 		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
-		
-		
-		if (_verbosity >= 1) {
+		if (_verbosity >= 2) {
 			System.out.println("Default Minimum Partions: " + jsc.defaultMinPartitions());
 			System.out.println("Default Parallelism: " + jsc.defaultParallelism());
 		}
 				
-		JavaRDD<String> json_list_data = jsc.textFile(_json_list_filename,6).cache();
+		JavaRDD<String> json_list_data = jsc.textFile(_json_list_filename,NUM_PARTITIONS).cache();
 
-		JavaRDD<String> json_ids = json_list_data.flatMap(new PagedJSON(_input_dir,_verbosity));
+		PagedJSON paged_json = new PagedJSON(_input_dir,_output_dir,_verbosity);
+		JavaRDD<String> json_ids = json_list_data.flatMap(paged_json).cache();
 
+		json_ids.saveAsTextFile("foo");
 
-		//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"); }
-		}).count();
-
-		System.out.println("#### Lines with a: " + numAs + ", lines with b: " + numBs);
-		 */
 		long num_ids = json_ids.count();
 		System.out.println("");
@@ -81,5 +63,5 @@
 
 		//.withType(Integer.class)
-
+/*
 		options.addOption(OptionBuilder.withLongOpt("verbosity")
 				.withDescription("Set to control the level of debugging output [0=none, 1=some, 2=lots]")
@@ -88,11 +70,18 @@
 				.isRequired(false)
 				.create());
-
+*/
 		//Option num_cores_opt = new Option("n", "num-cores", true, "Number of cores to use");
 		//num_cores_opt.setRequired(false);
 		//options.addOption(num_cores_opt);
 
+		Option verbosity_opt = new Option("v", "verbosity", true, 
+				"Set to control the level of debugging output [0=none, 1=some, 2=lots]");
+		verbosity_opt.setRequired(false);
+		options.addOption(verbosity_opt);
+		
 		//CommandLineParser parser = new DefaultParser(); // 1.3 and above
-		CommandLineParser parser = new GnuParser();
+		
+		// need to work with CLI v1.2 as this is the JAR that is bundled with Hadoop/Spark 
+		CommandLineParser parser = new GnuParser(); 
 		HelpFormatter formatter = new HelpFormatter();
 		CommandLine cmd;
@@ -112,10 +101,7 @@
 
 		//cmd.hasOption("json-filelist")
+		
 		String verbosity_str = cmd.getOptionValue("verbosity","0");
 		int verbosity = Integer.parseInt(verbosity_str);
-
-		//System.out.println(inputFilePath);
-		//System.out.println(outputFilePath);
-
 
 		String[] filtered_args = cmd.getArgs();
@@ -129,9 +115,4 @@
 		String output_dir = filtered_args[2];
 
-
-		//String json_list_filename = cmd.getArgs()[0]; // args[0];
-		//String json_list_filename = args[0];
-		//int num_cores = 2;
-
 		PrepareForIngest prep_for_ingest = new PrepareForIngest(input_dir,json_list_filename,output_dir,verbosity);
 		prep_for_ingest.exec();
