Changeset 31251 for other-projects

Timestamp:

2016-12-19T15:13:52+13:00 (7 years ago)

Author:

davidb

Message:

Code tidy up. Timed experiment showed sorting by key with num_partitions significantly faster on 1000 slice

File:

• other-projects/hathitrust/wcsa/extracted-features-solr/trunk/solr-ingest/src/main/java/org/hathitrust/extractedfeatures/ProcessForWhitelist.java (modified) (5 diffs)

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

@@ -60 +60 @@
         JavaSparkContext jsc = new JavaSparkContext(conf);
 
-        /*
-        if (_verbosity >= 2) {
-            System.out.println("Default Minimum Partions: " + jsc.defaultMinPartitions());
-            System.out.println("Default Parallelism: " + jsc.defaultParallelism());
-        }
-            */
-        
         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();
-
+        json_list_data.setName("JSON-file-list");
+        
         long num_volumes = json_list_data.count();
         double per_vol = 100.0/(double)num_volumes;
@@ -77 +71 @@
         DoubleAccumulator per_vol_progress_accum = jsc.sc().doubleAccumulator("Per Volume Progress Percent");
         
-        //String strict_file_io_str = System.getProperty("wcsa-ef-ingest.strict-file-io","true");
         boolean strict_file_io = Boolean.getBoolean("wcsa-ef-ingest.strict-file-io");
                 
@@ -84 +77 @@
                                      per_vol_progress_accum,per_vol,
                                      strict_file_io);
-        JavaRDD<String> words = json_list_data.flatMap(paged_solr_wordfreq_flatmap); // .cache() *****
-        
+        JavaRDD<String> words = json_list_data.flatMap(paged_solr_wordfreq_flatmap); 
+        words.setName("tokenized-words");
         
         JavaPairRDD<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {
             public Tuple2<String, Integer> call(String s) { return new Tuple2<String, Integer>(s, 1); }
         });
-
+        pairs.setName("single-word-count");
+        
         JavaPairRDD<String, Integer> counts = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
             public Integer call(Integer a, Integer b) { return a + b; }
         });
-
-        //counts.map(lambda (x,y): (y,x));
-        
-        
-        JavaPairRDD<Integer, String> swappedPair = counts.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
+        counts.setName("word-frequency");
+        
+        JavaPairRDD<Integer, String> swapped_pair = counts.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
                @Override
                public Tuple2<Integer, String> call(Tuple2<String, Integer> item) throws Exception {
@@ -105 +97 @@
 
             });
-
-        //JavaPairRDD<Integer, String> sorted_swapped_pair = swappedPair.sortByKey(false,num_partitions);
-        JavaPairRDD<Integer, String> sorted_swapped_pair = swappedPair.sortByKey(false,1);
-
+        swapped_pair.setName("frequency-word-swap");
+        
+        JavaPairRDD<Integer, String> sorted_swapped_pair = swapped_pair.sortByKey(false,num_partitions);
+        
+        sorted_swapped_pair.setName("descending-sorted-frequency-word");
+        
         JavaPairRDD<String, Integer> sorted_swaped_back_pair = sorted_swapped_pair.mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
                @Override
@@ -114 +108 @@
                    return item.swap();
                }
-
             });
-
-        /*
-
-        JavaPairRDD<String, Integer> sorted_counts 
-            = counts.map(item -> item.swap()) // interchanges position of entries in each tuple
-                .sortByKey(true, 1)         // 1st arg configures ascending sort, 2nd arg configures one task
-                .map(item -> item.swap());
-        
+        sorted_swaped_back_pair.setName("descending-word-frequency");
+        
+/*
+        JavaPairRDD<Integer, String> counts_swapped_pair 
+            = counts.mapToPair(item -> item.swap());
+        JavaPairRDD<Integer, String> counts_swapped_pair_sorted 
+            = counts_swapped_pair.sortByKey(true, 1);
+        JavaPairRDD<String, Integer> count_sorted = counts_swapped_pair_sorted.mapToPair(item -> item.swap());
         */
         
         
-        //sorted_counts.saveAsTextFile(_json_list_filename + ".out");
         String filename_root = _json_list_filename.replaceAll(".*/","").replaceAll("\\..*$","");
         String output_directory = "whitelist-" + filename_root + "-out";