Changeset 31271 for other-projects/hathitrust/wcsa

Timestamp:

2016-12-28T14:04:19+13:00 (7 years ago)

Author:

davidb

Message:

Updating of POS code to new files-per-partition paramater, plus some other related tweaks

Location:

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

Files:

• PerVolumePOSStreamFlatmap.java (modified) (6 diffs)
• ProcessForLangCount.java (modified) (2 diffs)
• ProcessForPOSCount.java (modified) (5 diffs)
• ProcessForSolrIngest.java (modified) (1 diff)

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

@@ -41 +41 @@
         JSONObject extracted_feature_record = JSONClusterFileIO.readJSONFile(full_json_file_in);
         
-        ArrayList<String> all_word_list = new ArrayList<String>();
+        ArrayList<String> all_pos_list = new ArrayList<String>();
         
         if (extracted_feature_record != null) {
@@ -62 +62 @@
             }
     
-            if (_verbosity >= 2) {
+            if (_verbosity >= 3) {
                 System.out.print("  Pages: ");
             }
@@ -70 +70 @@
                 String page_id = volume_id + "." + formatted_i;
 
-                if (_verbosity >= 2) {
+                if (_verbosity >= 3) {
                     if (i>0) {
                         System.out.print(", ");
@@ -78 +78 @@
 
                 if (i==(ef_page_count-1)) {
-                    if (_verbosity >= 2) {
+                    if (_verbosity >= 3) {
                         System.out.println();
                     }
@@ -87 +87 @@
                 if (ef_page != null) {
                     
-                    ArrayList<String> page_word_list = SolrDocJSON.generateTokenPosCountPOSLabels(volume_id, page_id, ef_page);                 
-                    all_word_list.addAll(page_word_list);
+                    ArrayList<String> page_pos_list = SolrDocJSON.generateTokenPosCountPOSLabels(volume_id, page_id, ef_page);                  
+                    all_pos_list.addAll(page_pos_list);
                 }
                 else {
@@ -109 +109 @@
         _progress_accum.add(_progress_step);
         
-        return all_word_list.iterator();
+        return all_pos_list.iterator();
     }
     

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

@@ -69 +69 @@
         
         //int num_partitions = Integer.getInteger("wcsa-ef-ingest.num-partitions", DEFAULT_NUM_PARTITIONS);
-        int files_per_partition = Integer.getInteger("wcsa-ef-ingest.num-partitions", DEFAULT_FILES_PER_PARTITION);
+        int files_per_partition = Integer.getInteger("wcsa-ef-ingest.files-per-partition", DEFAULT_FILES_PER_PARTITION);
         
-        
-        JavaRDD<String> json_list_data = jsc.textFile(_json_list_filename,files_per_partition).cache();
+        JavaRDD<String> json_list_data = jsc.textFile(_json_list_filename).cache();
         json_list_data.setName("JSON-file-list");
         
@@ -81 +80 @@
         
         JavaRDD<String> json_list_data_rp = json_list_data.repartition(num_partitions);
+        json_list_data_rp.setName("JSON-file-list--repartitioned");
         
         DoubleAccumulator per_vol_progress_accum = jsc.sc().doubleAccumulator("Per Volume Progress Percent");

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

@@ -27 +27 @@
     //   http://blog.cloudera.com/blog/2015/03/how-to-tune-your-apache-spark-jobs-part-2/
     
-    protected static final int DEFAULT_NUM_CORES = 6;
-    protected static final int DEFAULT_NUM_PARTITIONS = 3*DEFAULT_NUM_CORES; 
+    //protected static final int DEFAULT_NUM_CORES = 6;
+    //protected static final int DEFAULT_NUM_PARTITIONS = 3*DEFAULT_NUM_CORES; 
+    protected static final int DEFAULT_FILES_PER_PARTITION = 3000;
     
     protected String _input_dir;
@@ -53 +54 @@
     public void execPOSCount()
     {   
-        String spark_app_name = generateSparkAppName("Per Page");       
+        String spark_app_name = generateSparkAppName("Per Volume");     
         
         SparkConf conf = new SparkConf().setAppName(spark_app_name);
@@ -67 +68 @@
         }
         
-        int num_partitions = Integer.getInteger("wcsa-ef-ingest.num-partitions", DEFAULT_NUM_PARTITIONS);
-        JavaRDD<String> json_list_data = jsc.textFile(_json_list_filename,num_partitions).cache();
+        //int num_partitions = Integer.getInteger("wcsa-ef-ingest.num-partitions", DEFAULT_NUM_PARTITIONS);
+        int files_per_partition = Integer.getInteger("wcsa-ef-ingest.num-partitions", DEFAULT_FILES_PER_PARTITION);
+        
+        
+        JavaRDD<String> json_list_data = jsc.textFile(_json_list_filename).cache();
         json_list_data.setName("JSON-file-list");
         
@@ -74 +78 @@
         double per_vol = 100.0/(double)num_volumes;
 
+        int num_partitions = (int)(num_volumes/files_per_partition)+1;
+        JavaRDD<String> json_list_data_rp = json_list_data.repartition(num_partitions);
+        json_list_data_rp.setName("JSON-file-list--repartitioned");
+        
         DoubleAccumulator per_vol_progress_accum = jsc.sc().doubleAccumulator("Per Volume Progress Percent");
         
         boolean strict_file_io = Boolean.getBoolean("wcsa-ef-ingest.strict-file-io");
-        //boolean icu_tokenize = Boolean.getBoolean("wcsa-ef-ingest.icu-tokenize");
         
         PerVolumePOSStreamFlatmap paged_solr_posfreq_flatmap 
@@ -83 +90 @@
                                      per_vol_progress_accum,per_vol,
                                      strict_file_io);
-        JavaRDD<String> pos_list = json_list_data.flatMap(paged_solr_posfreq_flatmap); 
+        JavaRDD<String> pos_list = json_list_data_rp.flatMap(paged_solr_posfreq_flatmap); 
         pos_list.setName("pos-stream");
         

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

@@ -100 +100 @@
             
         //int num_partitions = Integer.getInteger("wcsa-ef-ingest.num-partitions", DEFAULT_NUM_PARTITIONS);
-        int files_per_partition = Integer.getInteger("wcsa-ef-ingest.num-partitions", DEFAULT_FILES_PER_PARTITION);
+        int files_per_partition = Integer.getInteger("wcsa-ef-ingest.files-per-partition", DEFAULT_FILES_PER_PARTITION);
         
         JavaRDD<String> json_list_data = jsc.textFile(_json_list_filename).cache();