/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * 
 *     http://www.apache.org/licenses/LICENSE-2.0
 *     
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.jena.hadoop.rdf.stats.jobs;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.SequenceFile.CompressionType;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.BZip2Codec;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.chain.ChainMapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.NLineInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.StringUtils;
import org.apache.jena.hadoop.rdf.io.input.QuadsInputFormat;
import org.apache.jena.hadoop.rdf.io.input.TriplesInputFormat;
import org.apache.jena.hadoop.rdf.io.input.TriplesOrQuadsInputFormat;
import org.apache.jena.hadoop.rdf.io.input.nquads.NQuadsInputFormat;
import org.apache.jena.hadoop.rdf.io.input.ntriples.NTriplesInputFormat;
import org.apache.jena.hadoop.rdf.io.output.nquads.NQuadsOutputFormat;
import org.apache.jena.hadoop.rdf.io.output.ntriples.NTriplesNodeOutputFormat;
import org.apache.jena.hadoop.rdf.io.output.ntriples.NTriplesOutputFormat;
import org.apache.jena.hadoop.rdf.mapreduce.KeyMapper;
import org.apache.jena.hadoop.rdf.mapreduce.RdfMapReduceConstants;
import org.apache.jena.hadoop.rdf.mapreduce.TextCountReducer;
import org.apache.jena.hadoop.rdf.mapreduce.characteristics.CharacteristicSetReducer;
import org.apache.jena.hadoop.rdf.mapreduce.characteristics.QuadCharacteristicSetGeneratingReducer;
import org.apache.jena.hadoop.rdf.mapreduce.characteristics.TripleCharacteristicSetGeneratingReducer;
import org.apache.jena.hadoop.rdf.mapreduce.count.NodeCountReducer;
import org.apache.jena.hadoop.rdf.mapreduce.count.QuadNodeCountMapper;
import org.apache.jena.hadoop.rdf.mapreduce.count.TripleNodeCountMapper;
import org.apache.jena.hadoop.rdf.mapreduce.count.datatypes.QuadDataTypeCountMapper;
import org.apache.jena.hadoop.rdf.mapreduce.count.datatypes.TripleDataTypeCountMapper;
import org.apache.jena.hadoop.rdf.mapreduce.count.namespaces.QuadNamespaceCountMapper;
import org.apache.jena.hadoop.rdf.mapreduce.count.namespaces.TripleNamespaceCountMapper;
import org.apache.jena.hadoop.rdf.mapreduce.count.positional.QuadGraphCountMapper;
import org.apache.jena.hadoop.rdf.mapreduce.count.positional.QuadObjectCountMapper;
import org.apache.jena.hadoop.rdf.mapreduce.count.positional.TripleObjectCountMapper;
import org.apache.jena.hadoop.rdf.mapreduce.filter.positional.QuadFilterByPredicateMapper;
import org.apache.jena.hadoop.rdf.mapreduce.filter.positional.TripleFilterByPredicateUriMapper;
import org.apache.jena.hadoop.rdf.mapreduce.group.QuadGroupBySubjectMapper;
import org.apache.jena.hadoop.rdf.mapreduce.group.TripleGroupBySubjectMapper;
import org.apache.jena.hadoop.rdf.mapreduce.transform.TriplesToQuadsConstantGraphMapper;
import org.apache.jena.hadoop.rdf.types.CharacteristicSetWritable;
import org.apache.jena.hadoop.rdf.types.NodeWritable;
import org.apache.jena.hadoop.rdf.types.QuadWritable;
import org.apache.jena.hadoop.rdf.types.TripleWritable;
import org.apache.jena.vocabulary.RDF ;

/**
 * Factory that can produce {@link Job} instances for computing various RDF
 * statistics
 * 
 * 
 * 
 */
public class JobFactory {

    /**
     * Private constructor prevents instantiation
     */
    private JobFactory() {
    }

    /**
     * Gets a job for computing node counts on RDF triple inputs
     * 
     * @param config
     *            Configuration
     * @param inputPaths
     *            Input paths
     * @param outputPath
     *            Output path
     * @return Job
     * @throws IOException
     */
    public static Job getTripleNodeCountJob(Configuration config, String[] inputPaths, String outputPath) throws IOException {
        Job job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Triples Node Usage Count");

        // Map/Reduce classes
        job.setMapperClass(TripleNodeCountMapper.class);
        job.setMapOutputKeyClass(NodeWritable.class);
        job.setMapOutputValueClass(LongWritable.class);
        job.setReducerClass(NodeCountReducer.class);

        // Input and Output
        job.setInputFormatClass(TriplesInputFormat.class);
        job.setOutputFormatClass(NTriplesNodeOutputFormat.class);
        FileInputFormat.setInputPaths(job, StringUtils.arrayToString(inputPaths));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        return job;
    }

    /**
     * Gets a job for computing node counts on RDF quad inputs
     * 
     * @param config
     *            Configuration
     * @param inputPaths
     *            Input paths
     * @param outputPath
     *            Output path
     * @return Job
     * @throws IOException
     */
    public static Job getQuadNodeCountJob(Configuration config, String[] inputPaths, String outputPath) throws IOException {
        Job job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Quads Node Usage Count");

        // Map/Reduce classes
        job.setMapperClass(QuadNodeCountMapper.class);
        job.setMapOutputKeyClass(NodeWritable.class);
        job.setMapOutputValueClass(LongWritable.class);
        job.setReducerClass(NodeCountReducer.class);

        // Input and Output
        job.setInputFormatClass(QuadsInputFormat.class);
        job.setOutputFormatClass(NTriplesNodeOutputFormat.class);
        FileInputFormat.setInputPaths(job, StringUtils.arrayToString(inputPaths));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        return job;
    }

    /**
     * Gets a job for computing node counts on RDF triple and/or quad inputs
     * 
     * @param config
     *            Configuration
     * @param inputPaths
     *            Input paths
     * @param outputPath
     *            Output path
     * @return Job
     * @throws IOException
     */
    public static Job getNodeCountJob(Configuration config, String[] inputPaths, String outputPath) throws IOException {
        Job job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Node Usage Count");

        // Map/Reduce classes
        job.setMapperClass(QuadNodeCountMapper.class);
        job.setMapOutputKeyClass(NodeWritable.class);
        job.setMapOutputValueClass(LongWritable.class);
        job.setReducerClass(NodeCountReducer.class);

        // Input and Output
        job.setInputFormatClass(TriplesOrQuadsInputFormat.class);
        job.setOutputFormatClass(NTriplesNodeOutputFormat.class);
        FileInputFormat.setInputPaths(job, StringUtils.arrayToString(inputPaths));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        return job;
    }
    
    public static Job getTripleGraphSizesJob(Configuration config, String[] inputPaths, String outputPath) throws IOException {
        Job job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Triples Graph Sizes");
        
        // Map/Reduce classes
        ChainMapper.addMapper(job, TriplesToQuadsConstantGraphMapper.class, LongWritable.class, TripleWritable.class, LongWritable.class, QuadWritable.class, config);
        ChainMapper.addMapper(job, QuadGraphCountMapper.class, LongWritable.class, QuadWritable.class, NodeWritable.class, LongWritable.class, config);
        job.setMapOutputKeyClass(NodeWritable.class);
        job.setMapOutputValueClass(LongWritable.class);
        job.setReducerClass(NodeCountReducer.class);
        
        // Input and Output
        job.setInputFormatClass(TriplesInputFormat.class);
        job.setOutputFormatClass(NTriplesNodeOutputFormat.class);
        FileInputFormat.setInputPaths(job, StringUtils.arrayToString(inputPaths));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));
        
        return job;
    }
    
    public static Job getQuadGraphSizesJob(Configuration config, String[] inputPaths, String outputPath) throws IOException {
        Job job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Quads Graph Sizes");
        
        // Map/Reduce classes
        job.setMapperClass(QuadGraphCountMapper.class);
        job.setMapOutputKeyClass(NodeWritable.class);
        job.setMapOutputValueClass(LongWritable.class);
        job.setReducerClass(NodeCountReducer.class);
        
        // Input and Output
        job.setInputFormatClass(QuadsInputFormat.class);
        job.setOutputFormatClass(NTriplesNodeOutputFormat.class);
        FileInputFormat.setInputPaths(job, StringUtils.arrayToString(inputPaths));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));
        
        return job;
    }
    
    public static Job getGraphSizesJob(Configuration config, String[] inputPaths, String outputPath) throws IOException {
        Job job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Graph Sizes");
        
        // Map/Reduce classes
        job.setMapperClass(QuadGraphCountMapper.class);
        job.setMapOutputKeyClass(NodeWritable.class);
        job.setMapOutputValueClass(LongWritable.class);
        job.setReducerClass(NodeCountReducer.class);
        
        // Input and Output
        job.setInputFormatClass(TriplesOrQuadsInputFormat.class);
        job.setOutputFormatClass(NTriplesNodeOutputFormat.class);
        FileInputFormat.setInputPaths(job, StringUtils.arrayToString(inputPaths));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));
        
        return job;
    }

    /**
     * Gets a sequence of jobs that can be used to compute characteristic sets
     * for RDF triples
     * 
     * @param config
     *            Configuration
     * @param inputPaths
     *            Input paths
     * @param intermediateOutputPath
     *            Intermediate output path
     * @param outputPath
     *            Final output path
     * @return Sequence of jobs
     * @throws IOException
     */
    public static Job[] getTripleCharacteristicSetJobs(Configuration config, String[] inputPaths, String intermediateOutputPath,
            String outputPath) throws IOException {
        Job[] jobs = new Job[2];

        Job job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Triples Characteristic Set (Generation)");

        // Map/Reduce classes
        job.setMapperClass(TripleGroupBySubjectMapper.class);
        job.setMapOutputKeyClass(NodeWritable.class);
        job.setMapOutputValueClass(TripleWritable.class);
        job.setReducerClass(TripleCharacteristicSetGeneratingReducer.class);
        job.setOutputKeyClass(CharacteristicSetWritable.class);
        job.setOutputValueClass(NullWritable.class);

        // Input and Output
        job.setInputFormatClass(TriplesInputFormat.class);
        job.setOutputFormatClass(SequenceFileOutputFormat.class);
        FileInputFormat.setInputPaths(job, StringUtils.arrayToString(inputPaths));
        FileOutputFormat.setOutputPath(job, new Path(intermediateOutputPath));
        SequenceFileOutputFormat.setCompressOutput(job, true);
        FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class);
        SequenceFileOutputFormat.setOutputCompressionType(job, CompressionType.BLOCK);

        jobs[0] = job;

        job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Triples Characteristic Set (Reduction)");

        // Map/Reduce classes
        job.setMapperClass(KeyMapper.class);
        job.setMapOutputKeyClass(CharacteristicSetWritable.class);
        job.setMapOutputValueClass(CharacteristicSetWritable.class);
        job.setReducerClass(CharacteristicSetReducer.class);
        job.setOutputKeyClass(CharacteristicSetWritable.class);
        job.setOutputValueClass(CharacteristicSetWritable.class);

        // Input and Output
        job.setInputFormatClass(SequenceFileInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        FileInputFormat.setInputPaths(job, intermediateOutputPath);
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        jobs[1] = job;
        return jobs;
    }

    /**
     * Gets a sequence of jobs that can be used to compute characteristic sets
     * for RDF quads
     * 
     * @param config
     *            Configuration
     * @param inputPaths
     *            Input paths
     * @param intermediateOutputPath
     *            Intermediate output path
     * @param outputPath
     *            Final output path
     * @return Sequence of jobs
     * @throws IOException
     */
    public static Job[] getQuadCharacteristicSetJobs(Configuration config, String[] inputPaths, String intermediateOutputPath,
            String outputPath) throws IOException {
        Job[] jobs = new Job[2];

        Job job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Quads Characteristic Set (Generation)");

        // Map/Reduce classes
        job.setMapperClass(QuadGroupBySubjectMapper.class);
        job.setMapOutputKeyClass(NodeWritable.class);
        job.setMapOutputValueClass(QuadWritable.class);
        job.setReducerClass(QuadCharacteristicSetGeneratingReducer.class);
        job.setOutputKeyClass(CharacteristicSetWritable.class);
        job.setOutputValueClass(NullWritable.class);

        // Input and Output
        job.setInputFormatClass(QuadsInputFormat.class);
        job.setOutputFormatClass(SequenceFileOutputFormat.class);
        FileInputFormat.setInputPaths(job, StringUtils.arrayToString(inputPaths));
        FileOutputFormat.setOutputPath(job, new Path(intermediateOutputPath));
        SequenceFileOutputFormat.setCompressOutput(job, true);
        FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class);
        SequenceFileOutputFormat.setOutputCompressionType(job, CompressionType.BLOCK);

        jobs[0] = job;

        job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Quads Characteristic Set (Reduction)");

        // Map/Reduce classes
        job.setMapperClass(KeyMapper.class);
        job.setMapOutputKeyClass(CharacteristicSetWritable.class);
        job.setMapOutputValueClass(CharacteristicSetWritable.class);
        job.setReducerClass(CharacteristicSetReducer.class);
        job.setOutputKeyClass(CharacteristicSetWritable.class);
        job.setOutputValueClass(CharacteristicSetWritable.class);

        // Input and Output
        job.setInputFormatClass(SequenceFileInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        FileInputFormat.setInputPaths(job, intermediateOutputPath);
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        jobs[1] = job;
        return jobs;
    }

    /**
     * Gets a sequence of jobs that can be used to compute characteristic sets
     * for RDF triple and/or quad inputs
     * 
     * @param config
     *            Configuration
     * @param inputPaths
     *            Input paths
     * @param intermediateOutputPath
     *            Intermediate output path
     * @param outputPath
     *            Final output path
     * @return Sequence of jobs
     * @throws IOException
     */
    public static Job[] getCharacteristicSetJobs(Configuration config, String[] inputPaths, String intermediateOutputPath,
            String outputPath) throws IOException {
        Job[] jobs = new Job[2];

        Job job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Characteristic Set (Generation)");

        // Map/Reduce classes
        job.setMapperClass(QuadGroupBySubjectMapper.class);
        job.setMapOutputKeyClass(NodeWritable.class);
        job.setMapOutputValueClass(QuadWritable.class);
        job.setReducerClass(QuadCharacteristicSetGeneratingReducer.class);
        job.setOutputKeyClass(CharacteristicSetWritable.class);
        job.setOutputValueClass(NullWritable.class);

        // Input and Output
        job.setInputFormatClass(TriplesOrQuadsInputFormat.class);
        job.setOutputFormatClass(SequenceFileOutputFormat.class);
        FileInputFormat.setInputPaths(job, StringUtils.arrayToString(inputPaths));
        FileOutputFormat.setOutputPath(job, new Path(intermediateOutputPath));
        SequenceFileOutputFormat.setCompressOutput(job, true);
        FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class);
        SequenceFileOutputFormat.setOutputCompressionType(job, CompressionType.BLOCK);

        jobs[0] = job;

        job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Characteristic Set (Reduction)");

        // Map/Reduce classes
        job.setMapperClass(KeyMapper.class);
        job.setMapOutputKeyClass(CharacteristicSetWritable.class);
        job.setMapOutputValueClass(CharacteristicSetWritable.class);
        job.setReducerClass(CharacteristicSetReducer.class);
        job.setOutputKeyClass(CharacteristicSetWritable.class);
        job.setOutputValueClass(CharacteristicSetWritable.class);

        // Input and Output
        job.setInputFormatClass(SequenceFileInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        FileInputFormat.setInputPaths(job, intermediateOutputPath);
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        jobs[1] = job;
        return jobs;
    }

    /**
     * Gets a job for computing type counts on RDF triple inputs
     * 
     * @param config
     *            Configuration
     * @param inputPaths
     *            Input paths
     * @param intermediateOutputPath
     *            Path for intermediate output which will be all the type
     *            declaration triples present in the inputs
     * @param outputPath
     *            Output path
     * @return Job
     * @throws IOException
     */
    public static Job[] getTripleTypeCountJobs(Configuration config, String[] inputPaths, String intermediateOutputPath,
            String outputPath) throws IOException {
        Job[] jobs = new Job[2];

        Job job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Type Triples Extraction");

        // Map/Reduce classes
        job.getConfiguration().setStrings(RdfMapReduceConstants.FILTER_PREDICATE_URIS, RDF.type.getURI());
        job.setMapperClass(TripleFilterByPredicateUriMapper.class);
        job.setMapOutputKeyClass(LongWritable.class);
        job.setMapOutputValueClass(TripleWritable.class);

        // Input and Output Format
        job.setInputFormatClass(TriplesInputFormat.class);
        job.setOutputFormatClass(NTriplesOutputFormat.class);
        FileInputFormat.setInputPaths(job, StringUtils.arrayToString(inputPaths));
        FileOutputFormat.setOutputPath(job, new Path(intermediateOutputPath));

        jobs[0] = job;

        // Object Node Usage count job
        job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Triples Type Usage Count");

        // Map/Reduce classes
        job.setMapperClass(TripleObjectCountMapper.class);
        job.setMapOutputKeyClass(NodeWritable.class);
        job.setMapOutputValueClass(LongWritable.class);
        job.setReducerClass(NodeCountReducer.class);

        // Input and Output
        job.setInputFormatClass(NTriplesInputFormat.class);
        NLineInputFormat.setNumLinesPerSplit(job, 10000); // TODO Would be
                                                          // better if this was
                                                          // intelligently
                                                          // configured
        job.setOutputFormatClass(NTriplesNodeOutputFormat.class);
        FileInputFormat.setInputPaths(job, intermediateOutputPath);
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        jobs[1] = job;

        return jobs;
    }

    /**
     * Gets a job for computing type counts on RDF quad inputs
     * 
     * @param config
     *            Configuration
     * @param inputPaths
     *            Input paths
     * @param intermediateOutputPath
     *            Path for intermediate output which will be all the type
     *            declaration quads present in the inputs
     * @param outputPath
     *            Output path
     * @return Job
     * @throws IOException
     */
    public static Job[] getQuadTypeCountJobs(Configuration config, String[] inputPaths, String intermediateOutputPath,
            String outputPath) throws IOException {
        Job[] jobs = new Job[2];

        Job job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Type Quads Extraction");

        // Map/Reduce classes
        job.getConfiguration().setStrings(RdfMapReduceConstants.FILTER_PREDICATE_URIS, RDF.type.getURI());
        job.setMapperClass(QuadFilterByPredicateMapper.class);
        job.setMapOutputKeyClass(LongWritable.class);
        job.setMapOutputValueClass(QuadWritable.class);

        // Input and Output Format
        job.setInputFormatClass(QuadsInputFormat.class);
        job.setOutputFormatClass(NQuadsOutputFormat.class);
        FileInputFormat.setInputPaths(job, StringUtils.arrayToString(inputPaths));
        FileOutputFormat.setOutputPath(job, new Path(intermediateOutputPath));

        jobs[0] = job;

        // Object Node Usage count job
        job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Quads Type Usage Count");

        // Map/Reduce classes
        job.setMapperClass(QuadObjectCountMapper.class);
        job.setMapOutputKeyClass(NodeWritable.class);
        job.setMapOutputValueClass(LongWritable.class);
        job.setReducerClass(NodeCountReducer.class);

        // Input and Output
        job.setInputFormatClass(NQuadsInputFormat.class);
        NLineInputFormat.setNumLinesPerSplit(job, 10000); // TODO Would be
                                                          // better if this was
                                                          // intelligently
                                                          // configured
        job.setOutputFormatClass(NTriplesNodeOutputFormat.class);
        FileInputFormat.setInputPaths(job, intermediateOutputPath);
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        jobs[1] = job;

        return jobs;
    }

    /**
     * Gets a job for computing type counts on RDF triple and/or quad inputs
     * 
     * @param config
     *            Configuration
     * @param inputPaths
     *            Input paths
     * @param intermediateOutputPath
     *            Path for intermediate output which will be all the type
     *            declaration quads present in the inputs
     * @param outputPath
     *            Output path
     * @return Job
     * @throws IOException
     */
    public static Job[] getTypeCountJobs(Configuration config, String[] inputPaths, String intermediateOutputPath,
            String outputPath) throws IOException {
        Job[] jobs = new Job[2];

        Job job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Type Extraction");

        // Map/Reduce classes
        job.getConfiguration().setStrings(RdfMapReduceConstants.FILTER_PREDICATE_URIS, RDF.type.getURI());
        job.setMapperClass(QuadFilterByPredicateMapper.class);
        job.setMapOutputKeyClass(LongWritable.class);
        job.setMapOutputValueClass(QuadWritable.class);

        // Input and Output Format
        job.setInputFormatClass(TriplesOrQuadsInputFormat.class);
        job.setOutputFormatClass(NQuadsOutputFormat.class);
        FileInputFormat.setInputPaths(job, StringUtils.arrayToString(inputPaths));
        FileOutputFormat.setOutputPath(job, new Path(intermediateOutputPath));

        jobs[0] = job;

        // Object Node Usage count job
        job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Type Usage Count");

        // Map/Reduce classes
        job.setMapperClass(QuadObjectCountMapper.class);
        job.setMapOutputKeyClass(NodeWritable.class);
        job.setMapOutputValueClass(LongWritable.class);
        job.setReducerClass(NodeCountReducer.class);

        // Input and Output
        job.setInputFormatClass(NQuadsInputFormat.class);
        NLineInputFormat.setNumLinesPerSplit(job, 10000); // TODO Would be
                                                          // better if this was
                                                          // intelligently
                                                          // configured
        job.setOutputFormatClass(NTriplesNodeOutputFormat.class);
        FileInputFormat.setInputPaths(job, intermediateOutputPath);
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        jobs[1] = job;

        return jobs;
    }

    /**
     * Gets a job for computing literal data type counts on RDF triple inputs
     * 
     * @param config
     *            Configuration
     * @param inputPaths
     *            Input paths
     * @param outputPath
     *            Output path
     * @return Job
     * @throws IOException
     */
    public static Job getTripleDataTypeCountJob(Configuration config, String[] inputPaths, String outputPath) throws IOException {
        Job job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Triples Literal Data Type Usage Count");

        // Map/Reduce classes
        job.setMapperClass(TripleDataTypeCountMapper.class);
        job.setMapOutputKeyClass(NodeWritable.class);
        job.setMapOutputValueClass(LongWritable.class);
        job.setReducerClass(NodeCountReducer.class);

        // Input and Output
        job.setInputFormatClass(TriplesInputFormat.class);
        job.setOutputFormatClass(NTriplesNodeOutputFormat.class);
        FileInputFormat.setInputPaths(job, StringUtils.arrayToString(inputPaths));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        return job;
    }

    /**
     * Gets a job for computing literal data type counts on RDF quad inputs
     * 
     * @param config
     *            Configuration
     * @param inputPaths
     *            Input paths
     * @param outputPath
     *            Output path
     * @return Job
     * @throws IOException
     */
    public static Job getQuadDataTypeCountJob(Configuration config, String[] inputPaths, String outputPath) throws IOException {
        Job job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Quads Literal Data Type Usage Count");

        // Map/Reduce classes
        job.setMapperClass(QuadDataTypeCountMapper.class);
        job.setMapOutputKeyClass(NodeWritable.class);
        job.setMapOutputValueClass(LongWritable.class);
        job.setReducerClass(NodeCountReducer.class);

        // Input and Output
        job.setInputFormatClass(QuadsInputFormat.class);
        job.setOutputFormatClass(NTriplesNodeOutputFormat.class);
        FileInputFormat.setInputPaths(job, StringUtils.arrayToString(inputPaths));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        return job;
    }

    /**
     * Gets a job for computing literal data type counts on RDF triple and/or
     * quad inputs
     * 
     * @param config
     *            Configuration
     * @param inputPaths
     *            Input paths
     * @param outputPath
     *            Output path
     * @return Job
     * @throws IOException
     */
    public static Job getDataTypeCountJob(Configuration config, String[] inputPaths, String outputPath) throws IOException {
        Job job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Literal Data Type Usage Count");

        // Map/Reduce classes
        job.setMapperClass(QuadDataTypeCountMapper.class);
        job.setMapOutputKeyClass(NodeWritable.class);
        job.setMapOutputValueClass(LongWritable.class);
        job.setReducerClass(NodeCountReducer.class);

        // Input and Output
        job.setInputFormatClass(TriplesOrQuadsInputFormat.class);
        job.setOutputFormatClass(NTriplesNodeOutputFormat.class);
        FileInputFormat.setInputPaths(job, StringUtils.arrayToString(inputPaths));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        return job;
    }

    /**
     * Gets a job for computing literal data type counts on RDF triple inputs
     * 
     * @param config
     *            Configuration
     * @param inputPaths
     *            Input paths
     * @param outputPath
     *            Output path
     * @return Job
     * @throws IOException
     */
    public static Job getTripleNamespaceCountJob(Configuration config, String[] inputPaths, String outputPath) throws IOException {
        Job job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Triples Namespace Usage Count");

        // Map/Reduce classes
        job.setMapperClass(TripleNamespaceCountMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);
        job.setReducerClass(TextCountReducer.class);

        // Input and Output
        job.setInputFormatClass(TriplesInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        FileInputFormat.setInputPaths(job, StringUtils.arrayToString(inputPaths));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        return job;
    }

    /**
     * Gets a job for computing literal data type counts on RDF quad inputs
     * 
     * @param config
     *            Configuration
     * @param inputPaths
     *            Input paths
     * @param outputPath
     *            Output path
     * @return Job
     * @throws IOException
     */
    public static Job getQuadNamespaceCountJob(Configuration config, String[] inputPaths, String outputPath) throws IOException {
        Job job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Quads Namespace Usage Count");

        // Map/Reduce classes
        job.setMapperClass(QuadNamespaceCountMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);
        job.setReducerClass(TextCountReducer.class);

        // Input and Output
        job.setInputFormatClass(QuadsInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        FileInputFormat.setInputPaths(job, StringUtils.arrayToString(inputPaths));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        return job;
    }

    /**
     * Gets a job for computing literal data type counts on RDF triple and/or
     * quad inputs
     * 
     * @param config
     *            Configuration
     * @param inputPaths
     *            Input paths
     * @param outputPath
     *            Output path
     * @return Job
     * @throws IOException
     */
    public static Job getNamespaceCountJob(Configuration config, String[] inputPaths, String outputPath) throws IOException {
        Job job = Job.getInstance(config);
        job.setJarByClass(JobFactory.class);
        job.setJobName("RDF Namespace Usage Count");

        // Map/Reduce classes
        job.setMapperClass(QuadNamespaceCountMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);
        job.setReducerClass(TextCountReducer.class);

        // Input and Output
        job.setInputFormatClass(TriplesOrQuadsInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        FileInputFormat.setInputPaths(job, StringUtils.arrayToString(inputPaths));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        return job;
    }
}