Changeset 31359 for other-projects/hathitrust/wcsa/extracted-features-solr/trunk/solr-ingest

Timestamp:

2017-01-26T23:08:16+13:00 (7 years ago)

Author:

davidb

Message:

Changed over to use sequenceFiles as input

Location:

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

Files:

• PerVolumeCatalogLangSequenceFileMap.java (added)
• ProcessForCatalogLangCount.java (modified) (2 diffs)

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

@@ -44 +44 @@
     }
         
-    public void execCatalogLangCount()
+    public void execCatalogLangCountSparkDirect()
     {   
-        String spark_app_name = generateSparkAppName("Per Volume");     
+        String spark_app_name = generateSparkAppName("Spark-Direct + Per Volume");      
         
         SparkConf conf = new SparkConf().setAppName(spark_app_name);
         JavaSparkContext jsc = new JavaSparkContext(conf);
-
+        
         String filename_root = _json_list_filename.replaceAll(".*/","").replaceAll("\\..*$","");
         String output_directory = "catalog-lang-" + filename_root + "-out";
@@ -110 +110 @@
     }
 
+    public void execCatalogLangCount()
+    {   
+        
+        
+        String spark_app_name = generateSparkAppName("YARN Cluster + Per Volume");      
+        
+        SparkConf conf = new SparkConf().setAppName(spark_app_name);
+        JavaSparkContext jsc = new JavaSparkContext(conf);
+        jsc.hadoopConfiguration().set("io.compression.codec.bzip2.library", "java-builtin");
+        
+        String packed_sequence_path = "hdfs:///user/capitanu/data/packed-ef";
+
+        JavaPairRDD<String, String> inputRdd = jsc.sequenceFile(packed_sequence_path, String.class, String.class);
+        JavaRDD<String> jsonTextRdd = inputRdd.map(Tuple2::_2);
+
+        /*
+        jsonTextRdd.map(
+            jsonText -> "" // parse JSON text and do whatever ...
+        );
+            */
+        
+        boolean strict_file_io = Boolean.getBoolean("wcsa-ef-ingest.strict-file-io");
+        
+        PerVolumeCatalogLangSequenceFileMap volume_catalog_langfreq_map
+            = new PerVolumeCatalogLangSequenceFileMap(_input_dir,_verbosity,strict_file_io);
+        JavaRDD<String> catalog_lang_list = jsonTextRdd.map(volume_catalog_langfreq_map); 
+        //catalog_lang_list.persist(StorageLevel.MEMORY_AND_DISK());
+        catalog_lang_list.setName("catalog-lang-stream");
+    
+        
+        JavaPairRDD<String, Long> catalog_lang_pairs = catalog_lang_list.mapToPair(s -> new Tuple2<String, Long>(s, 1L));
+        catalog_lang_pairs.setName("single-catalog-lang-count");
+        
+        JavaPairRDD<String, Long> catalog_lang_counts = catalog_lang_pairs.reduceByKey((a, b) -> a + b);
+        catalog_lang_counts.setName("catalog-lang-frequency");
+        
+        JavaPairRDD<Long, String> catalog_lang_counts_swapped_pair
+            = catalog_lang_counts.mapToPair(item -> item.swap());
+        catalog_lang_counts_swapped_pair.setName("frequency-catalog-lang-swap");
+        
+        JavaPairRDD<Long, String> catalog_lang_counts_swapped_pair_sorted 
+            = catalog_lang_counts_swapped_pair.sortByKey(false);
+        catalog_lang_counts_swapped_pair_sorted.setName("descending-sorted-frequency-catalog-lang");
+        
+        JavaPairRDD<String, Long> catalog_lang_count_sorted 
+            = catalog_lang_counts_swapped_pair_sorted.mapToPair(item -> item.swap());
+        catalog_lang_count_sorted.setName("descending-catalog-lang-frequency");
+        
+        String filename_root = _json_list_filename.replaceAll(".*/","").replaceAll("\\..*$","");
+        String output_directory = "catalog-lang-" + filename_root + "-out";
+        catalog_lang_count_sorted.saveAsTextFile(output_directory);
+        jsc.close();
+    }
+    
+    
     
     public static void print_usage(HelpFormatter formatter, Options options)