Changeset 30937

Timestamp:

2016-10-26T13:44:38+13:00 (8 years ago)

Author:

davidb

Message:

Expanded set of ClusterFileIO methods

Location:

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

Files:

• PagedJSON.java (modified) (4 diffs)
• PrepareForIngest.java (modified) (6 diffs)

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

@@ -1 +1 @@
 package org.hathitrust;
 
-import java.io.BufferedInputStream;
 import java.io.BufferedReader;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.net.URI;
 import java.util.ArrayList;
 import java.util.Iterator;
 
-import org.apache.commons.compress.compressors.CompressorException;
-import org.apache.commons.compress.compressors.CompressorInputStream;
-import org.apache.commons.compress.compressors.CompressorStreamFactory;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FSDataInputStream;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
 import org.apache.spark.api.java.function.FlatMapFunction;
 import org.json.JSONArray;
@@ -42 +30 @@
     }
     
-    protected static BufferedInputStream getBufferedInputStream(String fileIn) 
-            throws IOException 
-    {
-        BufferedInputStream bis = null;
-        
-        if (fileIn.startsWith("hdfs://")) {
-            URI uri = URI.create (fileIn);
-            Configuration conf = new Configuration();
-            FileSystem file = FileSystem.get(uri, conf);
-            FSDataInputStream fin = file.open(new Path(uri));
-
-            bis = new BufferedInputStream(fin);
-        }
-        else {
-            // Trim 'file://' off the front
-            
-            String local_file_in = fileIn;
-            if (local_file_in.startsWith("file://")) {
-                local_file_in = fileIn.substring("file://".length());
-            }
-            FileInputStream fin = new FileInputStream(local_file_in);
-            bis = new BufferedInputStream(fin);
-        }
-
-        return bis;
-        
-    }
-    protected static BufferedReader getBufferedReaderForCompressedFile(String fileIn) 
-            throws IOException, CompressorException 
-    {
-        BufferedInputStream bis = getBufferedInputStream(fileIn);
-        CompressorInputStream comp_input = new CompressorStreamFactory().createCompressorInputStream(bis);
-        BufferedReader br = new BufferedReader(new InputStreamReader(comp_input,"UTF8"));
-        return br;
-    }
-    
     protected JSONObject readJSONFile(String filename)
     {
@@ -87 +39 @@
 
             String str;
-            BufferedReader br = getBufferedReaderForCompressedFile(_input_dir + "/" + filename);
+            BufferedReader br = ClusterFileIO.getBufferedReaderForCompressedFile(_input_dir + "/" + filename);
             while ((str = br.readLine()) != null) {
                 sb.append(str);
@@ -120 +72 @@
     }
 
-    public Iterator<String> call(String s) 
+    public Iterator<String> call(String json_file_in) 
     { 
-        JSONObject extracted_feature_record = readJSONFile(s);
+        JSONObject extracted_feature_record = readJSONFile(json_file_in);
+        
+        // Check output directory for volume exists, and create it if not
+        
         
         String id = extracted_feature_record.getString("id");

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

@@ -19 +19 @@
     private static final long serialVersionUID = 1L;
 
-    protected String _input_dir;
-    protected String _json_list_filename;
-    protected String _output_dir;
-    protected int    _verbosity;
-    
+    protected String _input_dir;
+    protected String _json_list_filename;
+    protected String _output_dir;
+    
+    protected int    _verbosity;
+
     public PrepareForIngest(String input_dir, String json_list_filename, String output_dir, int verbosity)
     {
@@ -34 +35 @@
     public void exec()
     {   
-        SparkConf conf = new SparkConf().setAppName("HTRC-EF: Prepare for Solr Ingest");
-        JavaSparkContext sc = new JavaSparkContext(conf);
+        String spark_app_name = "HathiTrust Extract Features: Prepare for Solr Ingest";
+        spark_app_name += "[" + _json_list_filename + "]";
 
-        JavaRDD<String> json_list_data = sc.textFile(_json_list_filename).cache();
+        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
+    
+                
+        JavaRDD<String> json_list_data = jsc.textFile(_json_list_filename).cache();
 
         JavaRDD<String> json_ids = json_list_data.flatMap(new PagedJSON(_input_dir));
-        
+
 
         //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"); }
@@ -51 +59 @@
 
         System.out.println("#### Lines with a: " + numAs + ", lines with b: " + numBs);
-    */
+         */
         long num_ids = json_ids.count();
         System.out.println("");
@@ -58 +66 @@
         System.out.println("############");
         System.out.println("");
-        
-        sc.close();
+
+        jsc.close();
     }
 
@@ -90 +98 @@
         catch (ParseException e) {
             System.err.println(e.getMessage());
-            //System.err.println("Usage: RUN.bat [options] json-file-list.txt input-dir output-dir");
-            formatter.printHelp("RUN.bash/RUN.bat [options] json-file-list.txt input-dir output-dir", options);
-            //System.err.println("  Where 'filename.txt' contains a list of JSON files, one per line,");
-            //System.err.println("  which use the HathiTrust Extracted Feature JSON format");
-
+            formatter.printHelp("RUN.bash [options] json-file-list.txt input-dir output-dir", options);
             System.exit(1);
             return;
@@ -110 +114 @@
 
 
+        String[] filtered_args = cmd.getArgs();
 
-        String[] filtered_args = cmd.getArgs();
-        
         if (filtered_args.length != 3) {
-                //System.err.println("Usage: RUN.bat [options] json-filelist.txt input-dir output-dir");
-                formatter.printHelp("RUN.bash/RUN.bat  [options] json-filelist.txt input-dir output-dir", options);
-
-            //System.err.println("Usage: RUN.bat [options] input-dir output-dir");
-            //System.err.println("  Where 'filename.txt' contains a list of JSON files, one per line,");
-            //System.err.println("  which use the HathiTrust Extracted Feature JSON format");
-                System.exit(1);
+            formatter.printHelp("RUN.bash [options] json-filelist.txt input-dir output-dir", options);
+            System.exit(1);
         }
         String json_list_filename = filtered_args[0];
         String input_dir  = filtered_args[1];
         String output_dir = filtered_args[2];
-        
+
 
         //String json_list_filename = cmd.getArgs()[0]; // args[0];