MarkLogic 9 Product Documentation

MarkLogic Connector for Hadoop Developer's Guide — Chapter 3

Job Client

The Job Client prepares a job for execution.When you submit a MapReduce job to Hadoop, the local JobClient:

1. Validates the job configuration.
2. Generates the input splits. See How Hadoop Partitions Map Input Data.
3. Copies the job resources (configuration, job JAR file, input splits) to a shared location, such as an HDFS directory, where it is accessible to the Job Tracker and Task Trackers.
4. Submits the job to the Job Tracker.

Job Tracker

The Job Tracker is responsible for scheduling jobs, dividing a job into map and reduce tasks, distributing map and reduce tasks among worker nodes, task failure recovery, and tracking the job status. Job scheduling and failure recovery are not discussed here; see the documentation for your Hadoop distribution or the Apache Hadoop MapReduce documentation.

When preparing to run a job, the Job Tracker:

1. Fetches input splits from the shared location where the Job Client placed the information.
2. Creates a map task for each split.
3. Assigns each map task to a Task Tracker (worker node).

The Job Tracker monitors the health of the Task Trackers and the progress of the job. As map tasks complete and results become available, the Job Tracker:

1. Creates reduce tasks up to the maximum enableed by the job configuration.
2. Assigns each map result partition to a reduce task.
3. Assigns each reduce task to a Task Tracker.

A job is complete when all map and reduce tasks successfully complete, or, if there is no reduce step, when all map tasks successfully complete.

Task Tracker

A Task Tracker manages the tasks of one worker node and reports status to the Job Tracker. Often, the Task Tracker runs on the associated worker node, but it is not required to be on the same host.

When the Job Tracker assigns a map or reduce task to a Task Tracker, the Task Tracker:

1. Fetches job resources locally.
2. Spawns a child JVM on the worker node to execute the map or reduce task.
3. Reports status to the Job Tracker.

The task spawned by the Task Tracker runs the job's map or reduce functions.

Map Task

The Hadoop MapReduce framework creates a map task to process each input split. The map task:

1. Uses the InputFormat to fetch the input data locally and create input key-value pairs.
2. Applies the job-supplied map function to each key-value pair.
3. Performs local sorting and aggregation of the results.
4. If the job includes a Combiner, runs the Combiner for further aggregation.
5. Stores the results locally, in memory and on the local file system.
6. Communicates progress and status to the Task Tracker.

Map task results undergo a local sort by key to prepare the data for consumption by reduce tasks. If a Combiner is configured for the job, it also runs in the map task. A Combiner consolidates the data in an application-specific way, reducing the amount of data that must be transferred to reduce tasks. For example, a Combiner might compute a local maximum value for a key and discard the rest of the values. The details of how map tasks manage, sort, and shuffle results are not covered here. See the documentation for your Hadoop distribution or the Apache Hadoop MapReduce documentation.

When a map task notifies the Task Tracker of completion, the Task Tracker notifies the Job Tracker. The Job Tracker then makes the results available to reduce tasks.

Reduce Task

The reduce phase aggregates the results from the map phase into final results. Usually, the final result set is smaller than the input set, but this is application dependent. The reduction is carried out by parallel reduce tasks. The reduce input keys and values need not have the same type as the output keys and values.

The reduce phase is optional. You may configure a job to stop after the map phase completes. For details, see Configuring a Map-Only Job.

Reduce is carried out in three phases, copy, sort, and merge. A reduce task:

1. Fetches job resources locally.
2. Enters the copy phase to fetch local copies of all the assigned map results from the map worker nodes.
3. When the copy phase completes, executes the sort phase to merge the copied results into a single sorted set of (key, value-list) pairs.
4. When the sort phase completes, executes the reduce phase, invoking the job-supplied reduce function on each (key, value-list) pair.
5. Saves the final results to the output destination, such as HDFS.

The input to a reduce function is key-value pairs where the value is a list of values sharing the same key. For example, if one map task produces a key-value pair (eat, 2) and another map task produces the pair (eat, 1), then these pairs are consolidated into (eat, (2, 1)) for input to the reduce function. If the purpose of the reduce phase is to compute a sum of all the values for each key, then the final output key-value pair for this input is (eat, 3). For a more complete example, see Example: Calculating Word Occurrences.

Output from the reduce phase is saved to the destination configured for the job, such as HDFS or MarkLogic Server. Reduce tasks use an OutputFormat subclass to record results. The Hadoop API provides OutputFormat subclasses for using HDFS as the output destination. The MarkLogic Connector for Hadoop provides OutputFormat subclasses for using a MarkLogic Server database as the destination. For a list of available subclasses, see OutputFormat Subclasses. The connector also provides classes for defining key and value types; see MarkLogic-Specific Key and Value Types.

Configuration Basics

Hadoop configuration is controlled by multiple layers of configuration files and property settings. You may set configuration properties in configuration files, programmatically, and on the command line. For details, see the documentation for your Hadoop distribution, or the Apache Hadoop MapReduce documentation at http://hadoop.apache.org.

Configuration properties can include the following:

• Hadoop MapReduce properties
• Application-specific properties, such as the properties defined by the MarkLogic Connector for Hadoop API in com.marklogic.mapreduce.MarkLogicConstants
• Job-specific properties. For example, the mapreduce.linkcount.baseuri property used by the LinkCountInDoc sample application

A MapReduce application must configure at least the following:

• Mapper: Define a subclass of org.apache.hadoop.mapreduce.Mapper, usually overriding at least the Map.map() method.
• InputFormat: Select a subclass of org.apache.hadoop.mapreduce.InputFormat and pass it to org.apache.hadoop.mapreduce.Job.setInputFormatClass.

If the job includes a reduce step, then the application must also configure the following:

• Reducer: Define a subclass of org.apache.hadoop.mapreduce.Reducer, usually overriding at least the Reducer.reduce() method.
• OutputFormat: Select a subclass of org.apache.hadoop.mapreduce.OutputFormat and pass it to org.apache.hadoop.mapreduce.Job.setOutputFormatClass.
• Output key and value types for the map and reduce phases. Set the key and value types appropriate for your InputFormat and OutputFormat subclasses using the Job API functions, such as org.apache.hadoop.mapreduce.Job.setMapOutputKeyClass.

For details about configuring MapReduce jobs, see the documentation for your Hadoop distribution. For details about connector-specific configuration options, see Using MarkLogic Server for Input and Using MarkLogic Server for Output.

Setting Properties in a Configuration File

Configuration files are best suited for static configuration properties. By default, Apache Hadoop looks for configuration files in $HADOOP_CONF_DIR. You may override this location on the hadoop command line. Consult the documentation for your Hadoop distribution for the proper location, or see the Apache Hadoop Commands Guide at http://hadoop.apache.org.

Job configuration files are XML files with the following layout:

<?xml version=1.0?>
<?xml-stylesheet type=text/xsl href=configuration.xsl?>
<configuration>
  <property>
    <name>the.property.name</name>
    <value>the.property.value</value>
  </property>
  <property>...</property>
</configuration>

Setting Properties Using the Hadoop API

You can use the Apache Hadoop API to set properties that cannot be set in a configuration file or that have dynamically calculated values. For example, you can set the InputFormat class by calling org.apache.hadoop.mapreduce.Job.setInputFormatClass.

Set properties prior to submitting the job. That is, prior to calling org.apache.mapreduce.Job.submit or org.apache.mapreduce.Job.waitForCompletion.

To set an arbitrary property programmatically, use the org.apache.hadoop.conf.Configuration API. This API includes methods for setting property values to string, boolean, numeric, and class types. Set the properties prior to starting the job. For example, to set the MarkLogic Connector for Hadoop marklogic.mapreduce.input.documentselector property, at runtime, you can do the following:

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.Job;
import com.marklogic.mapreduce.MarkLogicConstants;
...
public class myJob { ...
  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    String[] otherArgs = 
        new GenericOptionsParser(conf, args).getRemainingArgs();
    Job job = new Job(conf);

    // Build up the document selectory dynamically...
    String mySelector = ...;

    conf = job.getConfiguration();
    conf.set(MarkLogicConstants.DOCUMENT_SELECTOR, mySelector);
    ...
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

Setting Properties on the Command Line

You can set properties on the hadoop command line using the -D command line option. For example:

$ hadoop -Dproperty_name=value jar myJob.jar myClass

Configuring a Map-Only Job

Some jobs can complete all their work during the map phase. For example, the ContentLoader sample application, which loads documents from HDFS into MarkLogic Server, performs document loading in the map tasks and then stops without using reduce tasks. See ContentLoader.

To stop a job after the map completes, set the number of reduce tasks to zero by setting the property mapred.reduce.tasks to 0 in your job configuration file, or by calling org.apache.hadoop.Configuration.setNumReduceTasks(0) in your application.