Index: /other-projects/hathitrust/wcsa/extracted-features-solr/trunk/solr-ingest/src/main/java/org/hathitrust/extractedfeatures/PerVolumeCatalogLangSequenceFileMap.java
===================================================================
--- /other-projects/hathitrust/wcsa/extracted-features-solr/trunk/solr-ingest/src/main/java/org/hathitrust/extractedfeatures/PerVolumeCatalogLangSequenceFileMap.java	(revision 31359)
+++ /other-projects/hathitrust/wcsa/extracted-features-solr/trunk/solr-ingest/src/main/java/org/hathitrust/extractedfeatures/PerVolumeCatalogLangSequenceFileMap.java	(revision 31359)
@@ -0,0 +1,66 @@
+package org.hathitrust.extractedfeatures;
+
+import java.io.IOException;
+import org.apache.spark.api.java.function.Function;
+import org.json.JSONObject;
+
+class PerVolumeCatalogLangSequenceFileMap implements Function<String, String>
+{
+	private static final long serialVersionUID = 1L;
+	
+	protected String _input_dir;
+	protected int    _verbosity;
+	
+	boolean _strict_file_io;
+	
+	public PerVolumeCatalogLangSequenceFileMap(String input_dir, int verbosity, boolean strict_file_io)
+	{
+		_input_dir  = input_dir;
+		_verbosity  = verbosity;
+		
+		_strict_file_io = strict_file_io;
+	}
+	
+	public String call(String json_text) throws IOException
+	{
+		String catalog_lang = null;
+		 
+		try {
+			JSONObject extracted_feature_record  = new JSONObject(json_text);
+
+			String volume_id = extracted_feature_record.getString("id");
+
+			JSONObject ef_metadata = extracted_feature_record.getJSONObject("metadata");
+
+			if (_verbosity >= 1) {
+				System.out.println("Processing: " + volume_id);
+			}
+
+			if (ef_metadata != null) {
+				String ef_catalog_language = ef_metadata.getString("language");
+				if (!ef_catalog_language.equals("")) {
+
+					catalog_lang = ef_catalog_language;
+				}
+				else {
+					System.err.println("No catalog 'language' metadata => Skipping id: " + volume_id);
+				}
+			}
+			else {
+				System.err.println("No 'metadata' section in JSON file => Skipping id: " + volume_id);
+			}
+
+		}
+		catch (Exception e) {
+			if (_strict_file_io) {
+				throw e;
+			}
+			else {
+				e.printStackTrace();
+			}
+		}
+		
+		return catalog_lang;
+	}
+}
+
Index: /other-projects/hathitrust/wcsa/extracted-features-solr/trunk/solr-ingest/src/main/java/org/hathitrust/extractedfeatures/ProcessForCatalogLangCount.java
===================================================================
--- /other-projects/hathitrust/wcsa/extracted-features-solr/trunk/solr-ingest/src/main/java/org/hathitrust/extractedfeatures/ProcessForCatalogLangCount.java	(revision 31358)
+++ /other-projects/hathitrust/wcsa/extracted-features-solr/trunk/solr-ingest/src/main/java/org/hathitrust/extractedfeatures/ProcessForCatalogLangCount.java	(revision 31359)
@@ -44,11 +44,11 @@
 	}
 		
-	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,4 +110,59 @@
 	}
 
+	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)
