Index: her-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/PerPageJSONForeach.java
===================================================================
--- /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/PerPageJSONForeach.java	(revision 31010)
+++ 	(revision )
@@ -1,81 +1,0 @@
-package org.hathitrust.extractedfeatures;
-
-import java.util.ArrayList;
-import java.util.Iterator;
-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"); }
-}
- */
-
-
-class PerPageJSONForeach implements VoidFunction<JSONObject> 
-//public class PagedJSON implements VoidFunction<String> 
-{
-	private static final long serialVersionUID = 1L;
-	
-	protected String _input_dir;
-	protected String _solr_url;
-	protected String _output_dir;
-	protected int    _verbosity;
-	
-	protected DoubleAccumulator _progress_accum;
-	protected double            _progress_step;
-	
-	
-	public PerPageJSONForeach(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 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("==================");
-			}
-			SolrDocJSON.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("==================");
-			}
-			SolrDocJSON.saveSolrDoc(solr_add_doc_json, _output_dir + "/" + output_json_bz2);
-		}
-	}
-	
-}
-
Index: /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/PerPageJSONMap.java
===================================================================
--- /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/PerPageJSONMap.java	(revision 31011)
+++ /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/PerPageJSONMap.java	(revision 31011)
@@ -0,0 +1,70 @@
+package org.hathitrust.extractedfeatures;
+
+import org.apache.spark.api.java.function.Function;
+import org.apache.spark.api.java.function.MapFunction;
+import org.apache.spark.util.DoubleAccumulator;
+import org.json.JSONObject;
+
+
+class PerPageJSONMap implements Function<JSONObject, String> 
+{
+	private static final long serialVersionUID = 1L;
+	
+	protected String _input_dir;
+	protected String _solr_url;
+	protected String _output_dir;
+	protected int    _verbosity;
+	
+	protected DoubleAccumulator _progress_accum;
+	protected double            _progress_step;
+	
+	
+	public PerPageJSONMap(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 String 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("==================");
+			}
+			SolrDocJSON.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("==================");
+			}
+			SolrDocJSON.saveSolrDoc(solr_add_doc_json, _output_dir + "/" + output_json_bz2);
+		}
+		
+		return 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 31010)
+++ /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/PerVolumeJSON.java	(revision 31011)
@@ -1,16 +1,4 @@
 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;
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 31010)
+++ /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/ProcessForSolrIngest.java	(revision 31011)
@@ -119,9 +119,9 @@
 		DoubleAccumulator progress_accum = jsc.sc().doubleAccumulator("Progress Percent");
 		
-		PerPageJSONFlatmap paged_json = new PerPageJSONFlatmap(_input_dir,_solr_url,_output_dir,_verbosity, progress_accum,per_vol);
-		JavaRDD<JSONObject> json_per_page_ids = json_list_data.flatMap(paged_json).cache();
-		
-		PerPageJSONForeach paged_json_foreach = new PerPageJSONForeach(_input_dir,_solr_url,_output_dir,_verbosity, progress_accum,per_vol);
-		json_per_page_ids.foreach(paged_json_foreach);
+		PerPageJSONFlatmap paged_solr_json_flatmap = new PerPageJSONFlatmap(_input_dir,_solr_url,_output_dir,_verbosity, progress_accum,per_vol);
+		JavaRDD<JSONObject> per_page_jsonobjects = json_list_data.flatMap(paged_solr_json_flatmap).cache();
+		
+		PerPageJSONMap paged_json_id_map = new PerPageJSONMap(_input_dir,_solr_url,_output_dir,_verbosity, progress_accum,per_vol);
+		JavaRDD<String> per_page_ids = per_page_jsonobjects.map(paged_json_id_map);
 
 /*
@@ -133,5 +133,5 @@
 */
 		
-		long num_page_ids = json_per_page_ids.count();
+		long num_page_ids = per_page_ids.count(); // trigger lazy eval of: flatmap:per-vol -> map:per-page
 		
 		System.out.println("");
Index: /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/__PerPageJSONForeach.java
===================================================================
--- /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/__PerPageJSONForeach.java	(revision 31011)
+++ /other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/__PerPageJSONForeach.java	(revision 31011)
@@ -0,0 +1,81 @@
+package org.hathitrust.extractedfeatures;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+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"); }
+}
+ */
+
+
+class __PerPageJSONForeach implements VoidFunction<JSONObject> 
+//public class PagedJSON implements VoidFunction<String> 
+{
+	private static final long serialVersionUID = 1L;
+	
+	protected String _input_dir;
+	protected String _solr_url;
+	protected String _output_dir;
+	protected int    _verbosity;
+	
+	protected DoubleAccumulator _progress_accum;
+	protected double            _progress_step;
+	
+	
+	public __PerPageJSONForeach(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 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("==================");
+			}
+			SolrDocJSON.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("==================");
+			}
+			SolrDocJSON.saveSolrDoc(solr_add_doc_json, _output_dir + "/" + output_json_bz2);
+		}
+	}
+	
+}
+
