Changeset 31028 for other-projects

Timestamp:

2016-11-02T14:17:45+13:00 (7 years ago)

Author:

davidb

Message:

Support for randonly choosing Solr endpoints added in

Location:

other-projects/hathitrust/wcsa/extracted-features-solr/trunk/solr-ingest

Files:

• _RUN.bash (modified) (1 diff)
• ef-solr.properties (modified) (1 diff)
• src/main/java/org/hathitrust/extractedfeatures/PerPageJSONMap.java (modified) (3 diffs)
• src/main/java/org/hathitrust/extractedfeatures/ProcessForSolrIngest.java (modified) (5 diffs)

other-projects/hathitrust/wcsa/extracted-features-solr/trunk/solr-ingest/_RUN.bash

@@ -82 +82 @@
 
 
-cmd="$cmd --verbosity 1 $input_dir $json_filelist $*"
+cmd="$cmd --properties ef-solr.properties $input_dir $json_filelist $*"
 
 echo "****"

other-projects/hathitrust/wcsa/extracted-features-solr/trunk/solr-ingest/ef-solr.properties

@@ -1 +1 @@
 
-#ef-solr.process-json-mode = per-volume
-ef-solr.process-json-mode = per-page
+#wcsa-ef-ingest.process-ef-json-mode = per-volume
+wcsa-ef-ingest.process-ef-json-mode = per-page
+
+wcsa-ef-ingest.solr-endpoint-ips = 10.11.0.53,10.11.0.54,10.11.0.55

other-projects/hathitrust/wcsa/extracted-features-solr/trunk/solr-ingest/src/main/java/org/hathitrust/extractedfeatures/PerPageJSONMap.java

@@ -1 +1 @@
 package org.hathitrust.extractedfeatures;
+
+import java.util.ArrayList;
 
 import org.apache.spark.api.java.function.Function;
 import org.apache.spark.api.java.function.MapFunction;
 import org.apache.spark.util.DoubleAccumulator;
+import org.apache.spark.util.LongAccumulator;
 import org.json.JSONObject;
 
@@ -12 +15 @@
     
     protected String _input_dir;
-    protected String _solr_url;
     protected String _output_dir;
     protected int    _verbosity;
     
-    protected DoubleAccumulator _progress_accum;
-    protected double            _progress_step;
+    protected final ArrayList<String> _solr_endpoints;
+    protected final int _solr_endpoints_len;
+    
+    protected LongAccumulator _progress_accum;
+    protected long            _progress_step;
     
     
-    public PerPageJSONMap(String input_dir, String solr_url, String output_dir, int verbosity, 
-                          DoubleAccumulator progress_accum, double progress_step)
+    public PerPageJSONMap(String input_dir, ArrayList<String> solr_endpoints, String output_dir, int verbosity, 
+                          LongAccumulator progress_accum, long progress_step)
     {
         _input_dir  = input_dir;
-        _solr_url   = solr_url;
         _output_dir = output_dir;
         _verbosity  = verbosity;
+        
+        _solr_endpoints   = solr_endpoints;
+        _solr_endpoints_len = _solr_endpoints.size();
         
         _progress_accum = progress_accum;
@@ -45 +52 @@
         }
         
-                    
-        if (_solr_url != null) {
+        String solr_url = null;
+        if (_solr_endpoints_len > 0) {
+            int random_choice = (int)(_solr_endpoints_len * Math.random());
+            solr_url = _solr_endpoints.get(random_choice);
+        }
+                
+        if (solr_url != null) {
             if ((_verbosity >=2) && (random_test)) {
                 System.out.println("==================");
-                System.out.println("Posting to: " + _solr_url);
+                System.out.println("Posting to: " + solr_url);
                 System.out.println("==================");
             }
-            SolrDocJSON.postSolrDoc(_solr_url, solr_add_doc_json);
+            SolrDocJSON.postSolrDoc(solr_url, solr_add_doc_json);
         }
 

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

@@ -6 +6 @@
 import java.io.IOException;
 import java.io.Serializable;
+import java.util.ArrayList;
+
 import org.apache.commons.cli.*;
 
 import org.apache.spark.api.java.*;
 import org.apache.spark.util.DoubleAccumulator;
+import org.apache.spark.util.LongAccumulator;
 import org.hathitrust.extractedfeatures.PerPageJSONFlatmap;
 import org.json.JSONObject;
@@ -103 +106 @@
         jsc.close();
     }
-
+    public ArrayList<String> extrapolateSolrEndpoints()
+    {
+        ArrayList<String> solr_endpoints = new ArrayList<String>();
+        
+        String solr_endpoint_ips = System.getProperty("wcsa-ef-ingest.solr-endpoint-ips",null);
+        if (solr_endpoint_ips != null) {
+            String [] ips = solr_endpoint_ips.split(",");
+            for (String ip : ips) {
+                String solr_endpoint = _solr_url.replaceFirst("//.*?:", "//"+ip+":");
+                solr_endpoints.add(solr_endpoint);
+            }
+        }
+        else {
+            if (_solr_url != null) {
+                solr_endpoints.add(_solr_url);
+            }
+        }
+
+        return solr_endpoints;
+    }
+    
     public void execPerPage()
     {   
@@ -129 +152 @@
         //long num_page_ids = per_page_jsonobjects.count(); // trigger lazy eval of: flatmap:per-vol
 
-        DoubleAccumulator per_page_progress_accum = jsc.sc().doubleAccumulator("Pages Processed");
+        LongAccumulator per_page_progress_accum = jsc.sc().longAccumulator("Pages Processed");
+        ArrayList<String> solr_endpoints = extrapolateSolrEndpoints();
         
         PerPageJSONMap paged_json_id_map 
-            = new PerPageJSONMap(_input_dir,_solr_url,_output_dir,_verbosity, per_page_progress_accum,1.0);
+            = new PerPageJSONMap(_input_dir,solr_endpoints,_output_dir,_verbosity, per_page_progress_accum,1);
         JavaRDD<String> per_page_ids = per_page_jsonobjects.map(paged_json_id_map);
 
@@ -219 +243 @@
 
         
-        String verbosity_str = cmd.getOptionValue("verbosity","0");
+        String verbosity_str = cmd.getOptionValue("verbosity","1");
         int verbosity = Integer.parseInt(verbosity_str);
 
@@ -269 +293 @@
             = new ProcessForSolrIngest(input_dir,json_list_filename,solr_url,output_dir,verbosity);
         
-        String process_json_mode = System.getProperty("ef-solr.process-json-mode","per-page");
-        if (process_json_mode.equals("per-volume")) { 
+        String process_ef_json_mode = System.getProperty("wcsa-ef-ingest.process-ef-json-mode","per-page");
+        if (process_ef_json_mode.equals("per-volume")) { 
             prep_for_ingest.execPerVolume();
         }