source: other-projects/hathitrust/solr-extracted-features/trunk/src/main/java/org/hathitrust/extractedfeatures/PerPageJSONMap.java@ 31013

package org.hathitrust.extractedfeatures;

import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.util.DoubleAccumulator;
import org.json.JSONObject;


class PerPageJSONMap implements Function<JSONObject, String> 
{
    private static final long serialVersionUID = 1L;
    
    protected String _input_dir;
    protected String _solr_url;
    protected String _output_dir;
    protected int    _verbosity;
    
    protected DoubleAccumulator _progress_accum;
    protected double            _progress_step;
    
    
    public PerPageJSONMap(String input_dir, String solr_url, String output_dir, int verbosity, 
                          DoubleAccumulator progress_accum, double progress_step)
    {
        _input_dir  = input_dir;
        _solr_url   = solr_url;
        _output_dir = output_dir;
        _verbosity  = verbosity;
        
        _progress_accum = progress_accum;
        _progress_step  = progress_step;
    }
    
    public String call(JSONObject solr_add_doc_json) 
    { 
        String output_json_bz2 = solr_add_doc_json.getString("filename_json_bz2");
        solr_add_doc_json.remove("filename_json_bz2");
        
        boolean random_test = (Math.random()>0.999); // every 1000
        
        if ((_verbosity >=2) && (random_test)) {
            System.out.println("==================");
            System.out.println("Sample output Solr add JSON [random test 1/1000]: " + solr_add_doc_json.toString());
            System.out.println("==================");
        }
        
                    
        if (_solr_url != null) {
            if ((_verbosity >=2) && (random_test)) {
                System.out.println("==================");
                System.out.println("Posting to: " + _solr_url);
                System.out.println("==================");
            }
            SolrDocJSON.postSolrDoc(_solr_url, solr_add_doc_json);
        }

        if (_output_dir != null) {
            if ((_verbosity >=2) && (random_test)) {
                System.out.println("==================");
                System.out.println("Saving to: " + _output_dir);
                System.out.println("==================");
            }
            SolrDocJSON.saveSolrDoc(solr_add_doc_json, _output_dir + "/" + output_json_bz2);
        }
        
        _progress_accum.add(_progress_step);
        
        return output_json_bz2;
    }
    
}