Changeset 30945 for other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/PrepareForIngest.java

Timestamp:

2016-10-26T15:37:24+13:00 (8 years ago)

Author:

davidb

Message:

Getting closer to writing out JSON files

File:

• other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/PrepareForIngest.java (modified) (7 diffs)

other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/PrepareForIngest.java

@@ -1 +1 @@
 package org.hathitrust;
 
-import java.io.IOException;
 import java.io.Serializable;
-import java.nio.charset.StandardCharsets;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.util.List;
-
 import org.apache.commons.cli.*;
 
 import org.apache.spark.api.java.*;
 import org.apache.spark.SparkConf;
-import org.apache.spark.api.java.function.Function;
 
 public class PrepareForIngest implements Serializable
@@ -19 +11 @@
     private static final long serialVersionUID = 1L;
 
+    public static final int NUM_PARTITIONS = 6; // default would appear to be 2
+    
     protected String _input_dir;
     protected String _json_list_filename;
@@ -40 +34 @@
         SparkConf conf = new SparkConf().setAppName(spark_app_name);
         JavaSparkContext jsc = new JavaSparkContext(conf);
-        //ClusterFileIO.init(_input_dir);
         
-        // Check output directory exists, and create it if not
-        
-        
-        if (_verbosity >= 1) {
+        if (_verbosity >= 2) {
             System.out.println("Default Minimum Partions: " + jsc.defaultMinPartitions());
             System.out.println("Default Parallelism: " + jsc.defaultParallelism());
         }
                 
-        JavaRDD<String> json_list_data = jsc.textFile(_json_list_filename,6).cache();
+        JavaRDD<String> json_list_data = jsc.textFile(_json_list_filename,NUM_PARTITIONS).cache();
 
-        JavaRDD<String> json_ids = json_list_data.flatMap(new PagedJSON(_input_dir,_verbosity));
+        PagedJSON paged_json = new PagedJSON(_input_dir,_output_dir,_verbosity);
+        JavaRDD<String> json_ids = json_list_data.flatMap(paged_json).cache();
 
+        json_ids.saveAsTextFile("foo");
 
-        //long numAs = json_list_data.filter(new ContainsA()).count();
-
-
-        /*
-        long numBs = json_list_data.filter(new Function<String, Boolean>() {
-            public Boolean call(String s) { return s.contains("b"); }
-        }).count();
-
-        System.out.println("#### Lines with a: " + numAs + ", lines with b: " + numBs);
-         */
         long num_ids = json_ids.count();
         System.out.println("");
@@ -81 +63 @@
 
         //.withType(Integer.class)
-
+/*
         options.addOption(OptionBuilder.withLongOpt("verbosity")
                 .withDescription("Set to control the level of debugging output [0=none, 1=some, 2=lots]")
@@ -88 +70 @@
                 .isRequired(false)
                 .create());
-
+*/
         //Option num_cores_opt = new Option("n", "num-cores", true, "Number of cores to use");
         //num_cores_opt.setRequired(false);
         //options.addOption(num_cores_opt);
 
+        Option verbosity_opt = new Option("v", "verbosity", true, 
+                "Set to control the level of debugging output [0=none, 1=some, 2=lots]");
+        verbosity_opt.setRequired(false);
+        options.addOption(verbosity_opt);
+        
         //CommandLineParser parser = new DefaultParser(); // 1.3 and above
-        CommandLineParser parser = new GnuParser();
+        
+        // need to work with CLI v1.2 as this is the JAR that is bundled with Hadoop/Spark 
+        CommandLineParser parser = new GnuParser(); 
         HelpFormatter formatter = new HelpFormatter();
         CommandLine cmd;
@@ -112 +101 @@
 
         //cmd.hasOption("json-filelist")
+        
         String verbosity_str = cmd.getOptionValue("verbosity","0");
         int verbosity = Integer.parseInt(verbosity_str);
-
-        //System.out.println(inputFilePath);
-        //System.out.println(outputFilePath);
-
 
         String[] filtered_args = cmd.getArgs();
@@ -129 +115 @@
         String output_dir = filtered_args[2];
 
-
-        //String json_list_filename = cmd.getArgs()[0]; // args[0];
-        //String json_list_filename = args[0];
-        //int num_cores = 2;
-
         PrepareForIngest prep_for_ingest = new PrepareForIngest(input_dir,json_list_filename,output_dir,verbosity);
         prep_for_ingest.exec();