Changeset 31001 for other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/ProcessForSolrIngest.java

Timestamp:

2016-10-30T23:51:07+13:00 (7 years ago)

Author:

davidb

Message:

Code to work per-volume and per-page

File:

• other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/ProcessForSolrIngest.java (modified) (6 diffs)

other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/ProcessForSolrIngest.java

@@ -7 +7 @@
 import org.apache.spark.util.DoubleAccumulator;
 import org.hathitrust.extractedfeatures.PagedJSON;
+import org.json.JSONObject;
 import org.apache.spark.SparkConf;
 
@@ -41 +42 @@
     }
 
-    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 +62 @@
         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 +110 @@
 */
         
-        //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 +133 @@
     }
 
+    
+    
+
     public static void print_usage(HelpFormatter formatter, Options options)
     {
@@ -174 +211 @@
         ProcessForSolrIngest prep_for_ingest 
             = new ProcessForSolrIngest(input_dir,json_list_filename,solr_url,output_dir,verbosity);
-        prep_for_ingest.exec();
+        prep_for_ingest.execPerVolume();
     }
 }