MarkLogic Connector for Hadoop Developer's Guide — Chapter 3

MapReduce Overview

Apache Hadoop MapReduce is a framework for processing large data sets in parallel across a Hadoop cluster. Data analysis uses a two step map and reduce process. The job configuration supplies map and reduce analysis functions and the Hadoop framework provides the scheduling, distribution, and parallelization services.

The top level unit of work in MapReduce is a job. A job usually has a map and a reduce phase, though the reduce phase can be omitted. For example, consider a MapReduce job that counts the number of times each word is used across a set of documents. The map phase counts the words in each document, then the reduce phase aggregates the per-document data into word counts spanning the entire collection.

During the map phase, the input data is divided into input splits for analysis by map tasks running in parallel across the Hadoop cluster. By default, the MapReduce framework gets input data from the Hadoop Distributed File System (HDFS). Using the MarkLogic Connector for Hadoop enables the framework to get input data from a MarkLogic Server instance. For details, see Map Task.

The reduce phase uses results from map tasks as input to a set of parallel reduce tasks. The reduce tasks consolidate the data into final results. By default, the MapReduce framework stores results in HDFS. Using the MarkLogic Connector for Hadoop enables the framework to store results in a MarkLogic Server instance. For details, see Reduce Task.

Although the reduce phase depends on output from the map phase, map and reduce processing is not necessarily sequential. That is, reduce tasks can begin as soon as any map task completes. It is not necessary for all map tasks to complete before any reduce task can begin.

MapReduce operates on key-value pairs. Conceptually, a MapReduce job takes a set of input key-value pairs and produces a set of output key-value pairs by passing the data through map and reduce functions. The map tasks produce an intermediate set of key-value pairs that the reduce tasks uses as input. The diagram below illustrates the progression from input key-value pairs to output key-value pairs at a high level:

Though each set of key-value pairs is homogeneous, the key-value pairs in each step need not have the same type. For example, the key-value pairs in the input set (KV1) can be (string, string) pairs, with the map phase producing (string, integer) pairs as intermediate results (KV2), and the reduce phase producing (integer, string) pairs for the final results (KV3). See Example: Calculating Word Occurrences.

The keys in the map output pairs need not be unique. Between the map processing and the reduce processing, a shuffle step sorts all map output values with the same key into a single reduce input (key, value-list) pair, where the value is a list of all values sharing the same key. Thus, the input to a reduce task is actually a set of (key, value-list) pairs.

The key and value types at each stage determine the interfaces to your map and reduce functions. Therefore, before coding a job, determine the data types needed at each stage in the map-reduce process. For example:

1. Choose the reduce output key and value types that best represents the desired outcome.
2. Choose the map input key and value types best suited to represent the input data from which to derive the final result.
3. Determine the transformation necessary to get from the map input to the reduce output, and choose the intermediate map output/reduce input key value type to match.

Control MapReduce job characteristics through configuration properties. The job configuration specifies:

• how to gather input
• the types of the input and output key-value pairs for each stage
• the map and reduce functions
• how and where to store the final results

For more information on job configuration, see Configuring a MapReduce Job. For information on MarkLogic specific configuration properties, see Input Configuration Properties and Output Configuration Properties.

Example: Calculating Word Occurrences

This example demonstrates the basic MapReduce concept by calculating the number of occurrence of each each word in a set of text files. For an in-depth discussion and source code for an equivalent example, see the Hadoop MapReduce tutorial at:

http://hadoop.apache.org/mapreduce/docs/current/mapred_tutorial.html

Recall that MapReduce input data is divided into input splits, and the splits are further divided into input key-value pairs. In this example, the input data set is the two documents, document1 and document2. The InputFormat subclass divides the data set into one split per document, for a total of 2 splits:

A (line number, text) key-value pair is generated for each line in an input document. The map function discards the line number and produces a per-line (word, count) pair for each word in the input line. The reduce phase produces (word, count) pairs representing aggregated word counts across all the input documents.

Given the input data shown above, the map-reduce progression for the example job is:

The output from the map phase contains multiple key-value pairs with the same key: The oats and eat keys appear twice. Recall that the MapReduce framework consolidates all values with the same key before entering the reduce phase, so the input to reduce is actually (key, values) pairs. Therefore, the full progression from map output, through reduce, to final results is:

Understanding the MapReduce Job Life Cycle

This section briefly sketches the life cycle of a MapReduce job and the roles of the primary actors in the life cycle. The full life cycle is much more complex. For details, refer to the documentation for your Hadoop distribution or the Apache Hadoop MapReduce documentation.

Though other configurations are possible, a common Hadoop cluster configuration is a single master node where the Job Tracker runs, and multiple worker nodes, each running a Task Tracker. The Job Tracker node can also be a worker node.

When the user submits a MapReduce job to Hadoop:

1. The local Job Client prepares the job for submission and hands it off to the Job Tracker.
2. The Job Tracker schedules the job and distributes the map work among the Task Trackers for parallel processing.
3. Each Task Tracker spawns a Map Task. The Job Tracker receives progress information from the Task Trackers.
4. As map results become available, the Job Tracker distributes the reduce work among the Task Trackers for parallel processing.
5. Each Task Tracker spawns a Reduce Task to perform the work. The Job Tracker receives progress information from the Task Trackers.

All map tasks do not have to complete before reduce tasks begin running. Reduce tasks can begin as soon as map tasks begin completing. Thus, the map and reduce steps often overlap.

How Hadoop Partitions Map Input Data

When you submit a job, the MapReduce framework divides the input data set into chunks called splits using the org.apache.hadoop.mapreduce.InputFormat subclass supplied in the job configuration. Splits are created by the local Job Client and included in the job information made available to the Job Tracker.

The JobTracker creates a map task for each split. Each map task uses a RecordReader provided by the InputFormat subclass to transform the split into input key-value pairs. The diagram below shows how the input data is broken down for analysis during the map phase:

The Hadoop API provides InputFormat subclasses for using HDFS as an input source. The MarkLogic Connector for Hadoop provides InputFormat subclasses for using MarkLogic Server as an input source. For a list of available MarkLogic-specific subclasses, see InputFormat Subclasses.

Configuring a MapReduce Job

This section covers the following topics:

• Configuration Basics
• Setting Properties in a Configuration File
• Setting Properties Using the Hadoop API
• Setting Properties on the Command Line
• Configuring a Map-Only Job

<?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>

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);
  }
}

$ hadoop -Dproperty_name=value jar myJob.jar myClass

Running a MapReduce Job

Use the hadoop jar command to execute a job. For example:

$ hadoop jar /path/myJob.jar my.package.Class options

For specific examples, see Using the Sample Applications. For details about the hadoop command line, consult the documentation for your Hadoop distribution or see the Apache Hadoop MapReduce Command Guide at http://hadoop.apache.org.

Viewing Job Status and Logs

A running job reports progress and errors to stdout and stderr. If Hadoop is configured for standalone mode, this output is all that is available. If Hadoop is configured in pseudo-distributed or fully distributed mode, logs are recorded in the Hadoop logs directory by default. The location of the logs directory is dependent on your Hadoop distribution. You can also view job status and results using the Job Tracker and NameNode web consoles.

This section assumes your Hadoop distribution uses the same logging mechanism as Apache Hadoop. Consult the documentation for your distribution for details.

Use the Job Tracker web console to view job status. By default, the Job Tracker console is available on port 50030. For example:

http://localhost:50030

Use the NameNode web console to browse HDFS, including job results, and to look at job related logs. By default, the NameNode console is available on port 50070. For example:

http://localhost:50070

To examine logs for a job:

1. Navigate to the NameNode console on port 50070.
2. Click the Namenode Logs link at the top of the page. A listing of the logs directory appears.
3. To check for errors, click on a Job Tracker log file, such as hadoop-your_username-jobtracker-your_hostname.log. The contents of the log file appears.
4. Click on your browser Back button to return to the log directory listing.
5. To view output output from your job, click on userlogs at the bottom of the logs directory listing. A user log listing appears.
6. Locate the run corresponding to your job by looking at the timestamps. A log directory is created for each map task and each reduce task.

   Map tasks have _m_ in the log directory name. Reduce tasks have _r_ in the log directory name. For example, attempt_201111020908_0001_m_000001_0/ represents a map task.

7. Click on the map or reduce task whose output you want to examine. A directory listing of the stdout, stderr, and syslog for the select task.
8. Click on the name of a log to see what the task output to that destination. For example, click stderr to view text written to stderr by the task.