ProActive MapReduce is a framework for execution of MapReduce jobs using the infrastructure provided by ProActive Scheduler and ProActive Resource Manager. The scheduler and resource manager provide fault tolerance, easy deployment and advanced scheduling capabilities not found in Hadoop. There is no distributed file system in ProActive though, so the DataSpaces mechanism is used for accessing input and writing output data.

ProActive MapReduce API is Hadoop-like. The user can define the ProActive MapReduce workflow as in Hadoop, but, internally, the ProActive MapReduce builds a ProActive job and submits it to ProActive Scheduler. Adapting existing Hadoop jobs to execute them using ProActive MapReduce framework is easy. No changes are required to the Mapper and Reducer classes; and only minimal amount of additional configuration is needed.

Let's start with a working Hadoop example (taken from Hadoop wiki):

import java.io.*;
import java.util.*;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class WordCount {

 public static class Map extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        StringTokenizer tokenizer = new StringTokenizer(line);
        while (tokenizer.hasMoreTokens()) {
            word.set(tokenizer.nextToken());
            context.write(word, one);
        }
    }
 }

 public static class Reduce extends Reducer<Text, IntWritable, Text, IntWritable> {

    public void reduce(Text key, Iterable<IntWritable> values, Context context)
      throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        context.write(key, new IntWritable(sum));
    }
 }

 public static void main(String[] args) throws Exception {

    // 1. instantiate Configuration and Job classes
    Configuration conf = new Configuration();
    Job job = new Job(conf, "wordcount");

    // 2. set the types of output keys and values produced by reducer
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);

    // 3. tell the job to use the previously defined Map and Reduce classes
    job.setMapperClass(Map.class);
    job.setReducerClass(Reduce.class);

    // 4. specify the formats of job input and output files
    job.setInputFormatClass(TextInputFormat.class);
    job.setOutputFormatClass(TextOutputFormat.class);

    // 5. specify the paths to input file and output directory; the
    // paths here are absolute HDFS paths
    FileInputFormat.addInputPath(job, new Path("/data/input/input_file"));
    FileOutputFormat.setOutputPath(job, new Path("/data/output/output_dir"));

    // 6. submit the job and wait for completion
    job.waitForCompletion(true);
 }

}

At the beginning of the WordCount class above, the Map and Reduce classes are defined, with map and reduce methods that do the actual work. The main method contains job creation, configuration and submission code.

In the following it is assumed that the environment variables SCHEDULER_HOME and EXAMPLE_DIR are set. $SCHEDULER_HOME should point to the directory where the scheduler is installed, and $EXAMPLE_DIR will be used for the code of the example, as well as job input and output files. The assumed values for those variables are listed below, feel free to adjust them to match your setup:

SCHEDULER_HOME=/home/user/proactive/scheduler
EXAMPLE_DIR=/home/user/mapreduce_example

We'll use variable reverences in the shell commands, and the corresponding values in java code and configuration files. We'll also use $EXAMPLE_DIR/data/input/input_file as an input path, and $EXAMPLE_DIR/data/output/output_dir as an output directory.

Here are the modifications necessary to be able to execute the above example in Proactive MapReduce:

Here is the full result of the modifications described above:

// filename: $MAPREDUCE_EXAMPLE/WordCount.java

import java.io.*;
import java.util.*;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import org.ow2.proactive.scheduler.ext.mapreduce.PAMapReduceJob;
import org.ow2.proactive.scheduler.ext.mapreduce.PAMapReduceJobConfiguration;
import org.ow2.proactive.scheduler.ext.mapreduce.exception.PAJobConfigurationException;

public class WordCount {

    public static class Map extends Mapper<LongWritable, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();

        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            StringTokenizer tokenizer = new StringTokenizer(line);
            while (tokenizer.hasMoreTokens()) {
                word.set(tokenizer.nextToken());
                context.write(word, one);
            }
        }
    }

    public static class Reduce extends Reducer<Text, IntWritable, Text, IntWritable> {

        public void reduce(Text key, Iterable<IntWritable> values, Context context)
            throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            context.write(key, new IntWritable(sum));
        }
    }

    public static void main(String[] args) throws Exception {

        // 1. instantiate Configuration and Job classes
        Configuration conf = new Configuration();
        Job job = new Job(conf, "wordcount");

        // 2. set the types of output keys and values produced by reducer
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 3. tell the job to use the previously defined Map and Reduce classes
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);

        // 4. specify the formats of job input and output files
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        // 5. specify the paths to input file and output directory;
        // the paths here are relative to the dataspaces declared in a
        // configuration file below
        FileInputFormat.addInputPath(job, new Path("input_file"));
        FileOutputFormat.setOutputPath(job, new Path("output_dir"));

        // 6. Instantiate PAMapReduceJobConfiguration using a path to
        // the configuration file
        File f = new File("/home/user/mapreduce_example/paMapReduceConfigurationProperties.property");
        PAMapReduceJobConfiguration pamrjc = new PAMapReduceJobConfiguration(f);

        // 7. Instantiate PAMapReduceJob using instances of Job and
        // PAMapReduceJobConfiguration as arguments
        PAMapReduceJob pamrj = null;
        try {
            pamrj = new PAMapReduceJob(job, pamrjc);
        } catch (PAJobConfigurationException e) {
            e.printStackTrace();
            System.exit(1);
        }

        // 8. submit the job to the scheduler
        if (pamrj.run())
            System.out.println("Submitted " + pamrj.getJobId());
        else

            System.out.println("Not submitted");
    }

}

To compile and run the example, take the following steps:

The ProActive MapReduce, when the user creates the ProActive MapReduce job and before that job is submitted to the ProActive Scheduler, internally builds a ProActive workflow made up of five tasks as the Figure 3.1, “ProActive MapReduce workflow” shows: SplitterPATask, MapperPATask, MapperJoinPATask, ReducerPATask and ReducerJoinPATask. The execution order of those tasks and their behavior is the following: the SplitterPATask creates the input splits from the input file. Each MapperPATask processes one and only one input split. To achieve that, MapperPATask is replicated with the replication factor equal to the number of created input splits. The MapperJoinPATask implements the join of the execution of the MapperPATask replicas. ReducerPATask starts its execution only after the MapperJoinPATask (and so all the MapperPATask replicas) ends. The ReducerPATask is replicated too. The number of replicas of the ReducerPATask is defined by the user. If the user does not define it, only one ReducerPATask is executed. The last executed task is the ReducerJoinPATask that implements the join of the ReducerPATask replicas.

Figure 3.1. ProActive MapReduce workflow

As was already mentioned in the Chapter 2, Tutorial: running Hadoop job using ProActive MapReduce, ProActive MapReduce configuration consists of 2 major steps:

Note: the ProActive MapReduce job is actually a ProActive workflow. Thus, many of its configuration properties are directly related to the ProActive workflow and to the tasks belonging to it, for instance, the cancelJobOnError attribute of a task.

There are two main restrictions related to the ProActive MapReduce: it requires Java 6 and it provides the support only for the Hadoop MapReduce 0.20.2 release and only for the Hadoop new MapReduce API (i.e., the Hadoop MapReduce job must be built using the classes defined in the package org.apache.hadoop.mapreduce, not org.apache.hadoop.mapred).

Moreover, the ProActive MapReduce Hadoop-like API implementation does not support many of the advanced Hadoop features, such as reporters, counters, compression, distributed cache, debug scripts and speculative execution. Those features, if specified during the configuration step, are ignored by the ProActive MapReduce.

To perform the tests we used the same group of hosts to run both Hadoop and ProActive MapReduce. The execution environment in which the benchamarks were performed is a cluster of 20 machines. The machines are identical and their characteristics are the following:

2 machines were reserved for running Hadoop JobTracker, HDFS NameNode, ProActive Scheduler and ProActive Resource Manager; the remaining 18 were used for dataNodes of the HDFS and execution of map and reduce tasks.

The version of JDK used for benchmarks is JDK 6 update 14.

Input files were generated using TPC-H dbgen utility, a database population program that generates files to be loaded into database tables. They consist of lines containing strings separated by "|" symbols. Of the several files generated by dbgen, we used only the file "lineitem.tbl".

On each of 18 machines 2 ProActive nodes were deployed for the executions of the ProActive MapReduce workflow. The main configuration parameters of the job were set as follows:

Each TaskTracker ^[1] was configured to run at most 6 map tasks during the map phase and 6 reduce tasks during the reduce phase. The size of input split was not configured and thus was chosen by Hadoop based on the HDFS block size (64 MB). The number of reducers was set to 54. Input files were uploaded to the HDFS with a replication factor of 3. We decided to include into the total execution time the time to upload the file to the HDFS to account for the fact that, unlike NFS used in ProActive MapReduce, HDFS is not a general purpose file system and thus a separate upload step is often required.

The results of the benchmark are given in the table below:

Legend: