Index: /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/JSONSolrTransform.java
===================================================================
--- /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/JSONSolrTransform.java	(revision 31000)
+++ /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/JSONSolrTransform.java	(revision 31001)
@@ -1,6 +1,14 @@
 package org.hathitrust.extractedfeatures;
 
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.URL;
 import java.util.Iterator;
 
+import org.apache.commons.compress.compressors.CompressorException;
 import org.json.JSONObject;
 
@@ -115,3 +123,69 @@
 	}
 
+	public static void saveSolrDoc(JSONObject solr_add_doc_json, String output_file_json_bz2)
+	{
+		try {
+			BufferedWriter bw = ClusterFileIO.getBufferedWriterForCompressedFile(output_file_json_bz2);
+			bw.write(solr_add_doc_json.toString());
+			bw.close();
+		} catch (IOException e) {
+			e.printStackTrace();
+		} catch (CompressorException e) {
+			e.printStackTrace();
+		}
+	}
+	
+	public static void postSolrDoc(String post_url, JSONObject solr_add_doc_json)
+	{
+		
+		//String curl_popen = "curl -X POST -H 'Content-Type: application/json'";
+		//curl_popen += " 'http://10.11.0.53:8983/solr/htrc-pd-ef/update'";
+		//curl_popen += " --data-binary '";
+		//curl_popen += "'"
+
+
+		try {
+			HttpURLConnection httpcon = (HttpURLConnection) ((new URL(post_url).openConnection()));
+			httpcon.setDoOutput(true);
+			httpcon.setRequestProperty("Content-Type", "application/json");
+			httpcon.setRequestProperty("Accept", "application/json");
+			httpcon.setRequestMethod("POST");
+			httpcon.connect();
+
+			byte[] outputBytes = solr_add_doc_json.toString().getBytes("UTF-8");
+			OutputStream os = httpcon.getOutputStream();
+			os.write(outputBytes);
+			os.close();
+			
+			
+			// Read response
+			StringBuilder sb = new StringBuilder();
+			BufferedReader in = new BufferedReader(new InputStreamReader(httpcon.getInputStream()));
+			String decodedString;
+			while ((decodedString = in.readLine()) != null) {
+				sb.append(decodedString);
+			}
+			in.close();
+
+			JSONObject solr_status_json = new JSONObject(sb.toString());
+			JSONObject response_header_json = solr_status_json.getJSONObject("responseHeader");
+			if (response_header_json != null) {
+				int status = response_header_json.getInt("status");
+				if (status != 0) {
+					System.err.println("Warning: POST request to " + post_url + " returned status " + status);
+					System.err.println("Full response was: " + sb);
+				}
+			}
+			else {
+				System.err.println("Failed response to Solr POST: " + sb);
+			}
+			
+			
+			
+		}
+		catch (Exception e) {
+			e.printStackTrace();
+		}
+		
+	}
 }
Index: /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/PagedJSON.java
===================================================================
--- /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/PagedJSON.java	(revision 31000)
+++ /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/PagedJSON.java	(revision 31001)
@@ -9,4 +9,5 @@
 import java.net.URL;
 import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.Set;
 
@@ -28,6 +29,6 @@
 
 
-//class PagedJSON implements FlatMapFunction<String, String> 
-public class PagedJSON implements VoidFunction<String> 
+class PagedJSON implements FlatMapFunction<String, JSONObject>, VoidFunction<JSONObject> 
+//public class PagedJSON implements VoidFunction<String> 
 {
 	private static final long serialVersionUID = 1L;
@@ -53,73 +54,8 @@
 	}
 	
-	public static void saveSolrDoc(JSONObject solr_add_doc_json, String output_file_json_bz2)
-	{
-		try {
-			BufferedWriter bw = ClusterFileIO.getBufferedWriterForCompressedFile(output_file_json_bz2);
-			bw.write(solr_add_doc_json.toString());
-			bw.close();
-		} catch (IOException e) {
-			e.printStackTrace();
-		} catch (CompressorException e) {
-			e.printStackTrace();
-		}
-	}
 	
-	public static void postSolrDoc(String post_url, JSONObject solr_add_doc_json)
-	{
-		
-		//String curl_popen = "curl -X POST -H 'Content-Type: application/json'";
-		//curl_popen += " 'http://10.11.0.53:8983/solr/htrc-pd-ef/update'";
-		//curl_popen += " --data-binary '";
-		//curl_popen += "'"
-
-
-		try {
-			HttpURLConnection httpcon = (HttpURLConnection) ((new URL(post_url).openConnection()));
-			httpcon.setDoOutput(true);
-			httpcon.setRequestProperty("Content-Type", "application/json");
-			httpcon.setRequestProperty("Accept", "application/json");
-			httpcon.setRequestMethod("POST");
-			httpcon.connect();
-
-			byte[] outputBytes = solr_add_doc_json.toString().getBytes("UTF-8");
-			OutputStream os = httpcon.getOutputStream();
-			os.write(outputBytes);
-			os.close();
-			
-			
-			// Read response
-			StringBuilder sb = new StringBuilder();
-			BufferedReader in = new BufferedReader(new InputStreamReader(httpcon.getInputStream()));
-			String decodedString;
-			while ((decodedString = in.readLine()) != null) {
-				sb.append(decodedString);
-			}
-			in.close();
-
-			JSONObject solr_status_json = new JSONObject(sb.toString());
-			JSONObject response_header_json = solr_status_json.getJSONObject("responseHeader");
-			if (response_header_json != null) {
-				int status = response_header_json.getInt("status");
-				if (status != 0) {
-					System.err.println("Warning: POST request to " + post_url + " returned status " + status);
-					System.err.println("Full response was: " + sb);
-				}
-			}
-			else {
-				System.err.println("Failed response to Solr POST: " + sb);
-			}
-			
-			
-			
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-		}
-		
-	}
 	
-	//public Iterator<String> call(String json_file_in) 
-	public void call(String json_file_in) 
+	public Iterator<JSONObject> call(String json_file_in) 
+	//public void call(String json_file_in) 
 	{ 
 		JSONObject extracted_feature_record = JSONClusterFileIO.readJSONFile(_input_dir + "/" + json_file_in);
@@ -148,5 +84,5 @@
 		ClusterFileIO.createDirectoryAll(_output_dir + "/" + page_json_dir);
 		
-		ArrayList<String> ids = new ArrayList<String>(ef_num_pages);
+		ArrayList<JSONObject> json_pages = new ArrayList<JSONObject>(ef_num_pages);
 		for (int i = 0; i < ef_page_count; i++) {
 			String formatted_i = String.format("page-%06d", i);
@@ -158,5 +94,5 @@
 			
 			String output_json_bz2 = page_json_dir +"/" + formatted_i + ".json.bz2";
-			ids.add(output_json_bz2);
+			//ids.add(output_json_bz2);
 			
 			if (i==0) {
@@ -165,34 +101,13 @@
 			
 			JSONObject ef_page = ef_pages.getJSONObject(i);
-
+			
 			if (ef_page != null) {
 				// Convert to Solr add form
 				JSONObject solr_add_doc_json = JSONSolrTransform.generateSolrDocJSON(volume_id, page_id, ef_page);
-
+				solr_add_doc_json.put("filename_json_bz2", output_json_bz2);
+				
+				json_pages.add(solr_add_doc_json);
 			
-				if ((_verbosity >=2) && (i==20)) {
-					System.out.println("==================");
-					System.out.println("Sample output Solr add JSON [page 20]: " + solr_add_doc_json.toString());
-					System.out.println("==================");
-				}
 				
-							
-				if (_solr_url != null) {
-					if ((_verbosity >=2) && (i==20)) {
-						System.out.println("==================");
-						System.out.println("Posting to: " + _solr_url);
-						System.out.println("==================");
-					}
-					postSolrDoc(_solr_url, solr_add_doc_json);
-				}
-
-				if (_output_dir != null) {
-					if ((_verbosity >=2) && (i==20)) {
-						System.out.println("==================");
-						System.out.println("Saving to: " + _output_dir);
-						System.out.println("==================");
-					}
-					saveSolrDoc(solr_add_doc_json, _output_dir + "/" + output_json_bz2);
-				}
 			}
 			else {
@@ -203,9 +118,44 @@
 		
 		
-		ids.add(volume_id);
+		//ids.add(volume_id);
 		_progress_accum.add(_progress_step);
 		
 		//return ids.iterator();
+		return json_pages.iterator();
 	}
+	
+	public void call(JSONObject solr_add_doc_json) 
+	{ 
+		String output_json_bz2 = solr_add_doc_json.getString("filename_json_bz2");
+		solr_add_doc_json.remove("filename_json_bz2");
+		
+		boolean random_test = (Math.random()>0.999); // every 1000
+		
+		if ((_verbosity >=2) && (random_test)) {
+			System.out.println("==================");
+			System.out.println("Sample output Solr add JSON [random test 1/1000]: " + solr_add_doc_json.toString());
+			System.out.println("==================");
+		}
+		
+					
+		if (_solr_url != null) {
+			if ((_verbosity >=2) && (random_test)) {
+				System.out.println("==================");
+				System.out.println("Posting to: " + _solr_url);
+				System.out.println("==================");
+			}
+			JSONSolrTransform.postSolrDoc(_solr_url, solr_add_doc_json);
+		}
+
+		if (_output_dir != null) {
+			if ((_verbosity >=2) && (random_test)) {
+				System.out.println("==================");
+				System.out.println("Saving to: " + _output_dir);
+				System.out.println("==================");
+			}
+			JSONSolrTransform.saveSolrDoc(solr_add_doc_json, _output_dir + "/" + output_json_bz2);
+		}
+	}
+	
 }
 
Index: /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/PerVolumeJSON.java
===================================================================
--- /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/PerVolumeJSON.java	(revision 31001)
+++ /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/PerVolumeJSON.java	(revision 31001)
@@ -0,0 +1,143 @@
+package org.hathitrust.extractedfeatures;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.Set;
+
+import org.apache.commons.compress.compressors.CompressorException;
+import org.apache.spark.api.java.function.FlatMapFunction;
+import org.apache.spark.api.java.function.VoidFunction;
+import org.apache.spark.util.DoubleAccumulator;
+import org.json.JSONArray;
+import org.json.JSONObject;
+
+/*
+class PagedJSON implements Function<String, Boolean> {
+
+	private static final long serialVersionUID = 1L;
+
+	public Boolean call(String s) { return s.contains("a"); }
+}
+ */
+
+
+public class PerVolumeJSON implements VoidFunction<String> 
+{
+	private static final long serialVersionUID = 1L;
+
+	protected String _input_dir;
+	protected String _solr_url;
+	protected String _output_dir;
+	protected int    _verbosity;
+	
+	DoubleAccumulator _progress_accum;
+	double            _progress_step;
+	
+	public PerVolumeJSON(String input_dir, String solr_url, String output_dir, int verbosity, 
+					 	 DoubleAccumulator progress_accum, double progress_step)
+	{
+		_input_dir  = input_dir;
+		_solr_url   = solr_url;
+		_output_dir = output_dir;
+		_verbosity  = verbosity;
+		
+		_progress_accum = progress_accum;
+		_progress_step  = progress_step;
+	}
+		
+	//public Iterator<String> call(String json_file_in) 
+	public void call(String json_file_in) 
+	{ 
+		JSONObject extracted_feature_record = JSONClusterFileIO.readJSONFile(_input_dir + "/" + json_file_in);
+		
+		String volume_id = extracted_feature_record.getString("id");
+		
+		//JSONObject ef_metadata = extracted_feature_record.getJSONObject("metadata");
+		//String title= ef_metadata.getString("title");
+		
+		JSONObject ef_features = extracted_feature_record.getJSONObject("features");
+		
+		
+		int ef_page_count = ef_features.getInt("pageCount");
+		
+		if (_verbosity >= 1) {
+			System.out.println("Processing: " + json_file_in);
+			System.out.println("  pageCount = " + ef_page_count);
+		}
+		
+		JSONArray ef_pages = ef_features.getJSONArray("pages");
+		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);
+		
+		//ArrayList<String> ids = new ArrayList<String>(ef_num_pages);
+		for (int i = 0; i < ef_page_count; i++) {
+			String formatted_i = String.format("page-%06d", i);
+			String page_id = volume_id + "." + formatted_i;
+		
+			if (_verbosity >= 2) {
+			  System.out.println("  Page: " + page_id);
+			}
+			
+			String output_json_bz2 = page_json_dir +"/" + formatted_i + ".json.bz2";
+			//ids.add(output_json_bz2); // ****
+			
+			if (i==0) {
+				System.out.println("Sample output JSON page file: " + output_json_bz2);
+			}
+			
+			JSONObject ef_page = ef_pages.getJSONObject(i);
+
+			if (ef_page != null) {
+				// Convert to Solr add form
+				JSONObject solr_add_doc_json = JSONSolrTransform.generateSolrDocJSON(volume_id, page_id, ef_page);
+
+			
+				if ((_verbosity >=2) && (i==20)) {
+					System.out.println("==================");
+					System.out.println("Sample output Solr add JSON [page 20]: " + solr_add_doc_json.toString());
+					System.out.println("==================");
+				}
+				
+							
+				if (_solr_url != null) {
+					if ((_verbosity >=2) && (i==20)) {
+						System.out.println("==================");
+						System.out.println("Posting to: " + _solr_url);
+						System.out.println("==================");
+					}
+					JSONSolrTransform.postSolrDoc(_solr_url, solr_add_doc_json);
+				}
+
+				if (_output_dir != null) {
+					if ((_verbosity >=2) && (i==20)) {
+						System.out.println("==================");
+						System.out.println("Saving to: " + _output_dir);
+						System.out.println("==================");
+					}
+					JSONSolrTransform.saveSolrDoc(solr_add_doc_json, _output_dir + "/" + output_json_bz2);
+				}
+			}
+			else {
+				System.err.println("Skipping: " + page_id);
+			}
+			
+		}
+		
+		
+		//ids.add(volume_id);
+		_progress_accum.add(_progress_step);
+		
+		//return ids.iterator();
+	}
+}
+
Index: /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/ProcessForSolrIngest.java
===================================================================
--- /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/ProcessForSolrIngest.java	(revision 31000)
+++ /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/ProcessForSolrIngest.java	(revision 31001)
@@ -7,4 +7,5 @@
 import org.apache.spark.util.DoubleAccumulator;
 import org.hathitrust.extractedfeatures.PagedJSON;
+import org.json.JSONObject;
 import org.apache.spark.SparkConf;
 
@@ -41,7 +42,7 @@
 	}
 
-	public void exec()
+	public void execPerVolume()
 	{	
-		String spark_app_name = "HathiTrust Extract Features: Prepare for Solr Ingest";
+		String spark_app_name = "[Per Volume] Extract Features: Process for Solr Ingest";
 		spark_app_name += " [" + _json_list_filename + "]";
 
@@ -61,10 +62,44 @@
 		DoubleAccumulator progress_accum = jsc.sc().doubleAccumulator("Progress Percent");
 		
+		PerVolumeJSON per_vol_json = new PerVolumeJSON(_input_dir,_solr_url,_output_dir,_verbosity, progress_accum,per_vol);
+
+		json_list_data.foreach(per_vol_json);
+		
+		long num_ids = num_volumes;
+		
+		System.out.println("");
+		System.out.println("############");
+		System.out.println("# Number of volume ids: " + num_ids);
+		System.out.println("############");
+		System.out.println("");
+
+		jsc.close();
+	}
+
+	public void execPerPage()
+	{	
+		String spark_app_name = "[Per Page] Extract Features: Process for Solr Ingest";
+		spark_app_name += " [" + _json_list_filename + "]";
+
+		SparkConf conf = new SparkConf().setAppName(spark_app_name);
+		JavaSparkContext jsc = new JavaSparkContext(conf);
+		
+		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,NUM_PARTITIONS).cache();
+
+		long num_volumes = json_list_data.count();
+		double per_vol = 100.0/(double)num_volumes;
+		
+		DoubleAccumulator progress_accum = jsc.sc().doubleAccumulator("Progress Percent");
+		
 		PagedJSON paged_json = new PagedJSON(_input_dir,_solr_url,_output_dir,_verbosity, progress_accum,per_vol);
-		//JavaRDD<String> json_ids = json_list_data.flatMap(paged_json).cache();
-
-		json_list_data.foreach(paged_json);
-
-	
+		JavaRDD<JSONObject> json_per_page_ids = json_list_data.flatMap(paged_json).cache();
+		
+		json_per_page_ids.foreach(paged_json);
+
 /*
 		System.out.println("");
@@ -75,10 +110,9 @@
 */
 		
-		//long num_ids = json_ids.count();
-		long num_ids = num_volumes;
-		
-		System.out.println("");
-		System.out.println("############");
-		System.out.println("# Number of page ids: " + num_ids);
+		long num_page_ids = json_per_page_ids.count();
+		
+		System.out.println("");
+		System.out.println("############");
+		System.out.println("# Number of page ids: " + num_page_ids);
 		System.out.println("############");
 		System.out.println("");
@@ -99,4 +133,7 @@
 	}
 
+	
+	
+
 	public static void print_usage(HelpFormatter formatter, Options options)
 	{
@@ -174,5 +211,5 @@
 		ProcessForSolrIngest prep_for_ingest 
 			= new ProcessForSolrIngest(input_dir,json_list_filename,solr_url,output_dir,verbosity);
-		prep_for_ingest.exec();
+		prep_for_ingest.execPerVolume();
 	}
 }
