Hadoop Quick Notes :: Part

From our previous exercises, we kicked off a Hadoop MapReduce process by:

Creating an instance of a Job object
Specifying the Mapper class
Specifying the Reducer class
Pointing the HDFS location to the input file(s)
Indicating the HDFS location to the output directory
And, finally submitting the MapReduce job

When a Job is submitted, it is put in a queue to be picked up and processed by the JobTracker.

Here is what the JobTracker does:

Checks for the existence of the output directory in HDFS. If exists, terminates with an error
Uses an implementation of org.apache.hadoop.mapreduce.InputFormat that validates the existence of the input directory in HDFS. If does not exist, terminates with an error
Uses the InputFormat to determine all the input file(s) to be processed under the specified HDFS input directory and returns a list of org.apache.hadoop.mapreduce.InputSplit (encapsulates the size of the data block as well as the list of nodes in the cluster where the data block resides)
Copies all the resources (Java jar, the configuration information, and the list of InputSplits) needed to execute the job in HDFS
Based on the information from each InputSplit, directs the appropriate TaskTracker closest to the input (data block) for starting the Mapper task. The number of Mapper tasks is controlled by the system property mapred.map.tasks (Default is 2)
Periodically gets a heartbeat from each of the TaskTrackers on the status of the Mapper tasks
Once all the Mapper tasks complete successfully, directs the available TaskTrackers to start the Reducer tasks. The number of Reducer tasks is controlled by the system property mapred.reduce.tasks (Default is 1)
Periodically gets a heartbeat from the TaskTrackers on the status of the Reducer tasks
Once the Reducer task finishes, the submitted Job is completed

The following Figure-1 illustrates the end-to-end picture of how it works:

Figure-1

The picture in Figure-1 is from the O'Reilly book: Hadoop, The Definitie Guide by Tom White.

In the above paragraph describing the JobTracker, we indicated few classes such as the InputFormat, InputSplit, etc. Let us elaborate on them in the following paragraph(s).

InputSplit

Used by the JobTracker
Is typically a data block from the HDFS with a byte-oriented view of the data
Encapsulates the size of the input data block as well as the list of nodes in the cluster where the data block(s) reside
Used as the input for the Mapper task
The default implementation for InputSplit is org.apache.hadoop.mapreduce.lib.input.FileSplit

The following is the source code for InputSplit taken as is from the hadoop-1.2.1:

Listing-1

/**
 * 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.hadoop.mapreduce;

import java.io.IOException;

import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.RecordReader;

/**
 * <code>InputSplit</code> represents the data to be processed by an 
 * individual {@link Mapper}. 
 *
 * <p>Typically, it presents a byte-oriented view on the input and is the 
 * responsibility of {@link RecordReader} of the job to process this and present
 * a record-oriented view.
 * 
 * @see InputFormat
 * @see RecordReader
 */
public abstract class InputSplit {
  /**
   * Get the size of the split, so that the input splits can be sorted by size.
   * @return the number of bytes in the split
   * @throws IOException
   * @throws InterruptedException
   */
  public abstract long getLength() throws IOException, InterruptedException;

  /**
   * Get the list of nodes by name where the data for the split would be local.
   * The locations do not need to be serialized.
   * @return a new array of the node nodes.
   * @throws IOException
   * @throws InterruptedException
   */
  public abstract 
    String[] getLocations() throws IOException, InterruptedException;
}

As can be inferred from Listing-1 above, InputSplit is an abstract class with no implementation.

The concrete implementation for InputSplit is the class FileSplit.

InputFormat

Used by the JobTracker
Validates the specified HDFS input directory
Based on the input file(s) to be processed, returns a list of org.apache.hadoop.mapreduce.InputSplit objects
Is a factory for org.apache.hadoop.mapreduce.RecordReader
The default implementation for InputFormat is org.apache.hadoop.mapreduce.lib.input.TextInputFormat. TextInputFormat is used for processing text file(s)

The following is the source code for InputFormat taken as is from the hadoop-1.2.1:

Listing-2

/**
 * 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.hadoop.mapreduce;

import java.io.IOException;
import java.util.List;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

/** 
 * <code>InputFormat</code> describes the input-specification for a 
 * Map-Reduce job. 
 * 
 * <p>The Map-Reduce framework relies on the <code>InputFormat</code> of the
 * job to:<p>
 * <ol>
 *   <li>
 *   Validate the input-specification of the job. 
 *   <li>
 *   Split-up the input file(s) into logical {@link InputSplit}s, each of 
 *   which is then assigned to an individual {@link Mapper}.
 *   </li>
 *   <li>
 *   Provide the {@link RecordReader} implementation to be used to glean
 *   input records from the logical <code>InputSplit</code> for processing by 
 *   the {@link Mapper}.
 *   </li>
 * </ol>
 * 
 * <p>The default behavior of file-based {@link InputFormat}s, typically 
 * sub-classes of {@link FileInputFormat}, is to split the 
 * input into <i>logical</i> {@link InputSplit}s based on the total size, in 
 * bytes, of the input files. However, the {@link FileSystem} blocksize of  
 * the input files is treated as an upper bound for input splits. A lower bound 
 * on the split size can be set via 
 * <a href="{@docRoot}/../mapred-default.html#mapred.min.split.size">
 * mapred.min.split.size</a>.</p>
 * 
 * <p>Clearly, logical splits based on input-size is insufficient for many 
 * applications since record boundaries are to respected. In such cases, the
 * application has to also implement a {@link RecordReader} on whom lies the
 * responsibility to respect record-boundaries and present a record-oriented
 * view of the logical <code>InputSplit</code> to the individual task.
 *
 * @see InputSplit
 * @see RecordReader
 * @see FileInputFormat
 */
public abstract class InputFormat<K, V> {

  /** 
   * Logically split the set of input files for the job.  
   * 
   * <p>Each {@link InputSplit} is then assigned to an individual {@link Mapper}
   * for processing.</p>
   *
   * <p><i>Note</i>: The split is a <i>logical</i> split of the inputs and the
   * input files are not physically split into chunks. For e.g. a split could
   * be <i>&lt;input-file-path, start, offset&gt;</i> tuple. The InputFormat
   * also creates the {@link RecordReader} to read the {@link InputSplit}.
   * 
   * @param context job configuration.
   * @return an array of {@link InputSplit}s for the job.
   */
  public abstract 
    List<InputSplit> getSplits(JobContext context
                               ) throws IOException, InterruptedException;
  
  /**
   * Create a record reader for a given split. The framework will call
   * {@link RecordReader#initialize(InputSplit, TaskAttemptContext)} before
   * the split is used.
   * @param split the split to be read
   * @param context the information about the task
   * @return a new record reader
   * @throws IOException
   * @throws InterruptedException
   */
  public abstract 
    RecordReader<K,V> createRecordReader(InputSplit split,
                                         TaskAttemptContext context
                                        ) throws IOException, 
                                                 InterruptedException;

}

As can be inferred from Listing-2 above, InputFormat is an abstract class with no implementation.

One of the concrete implementations of InputFormat for file based input is the class TextInputFormat.

The class TextInputFormat however extends the abstract class FileInputFormat.

The abstract class FileInputFormat provides the implementation for the method getSplits(), while the class TextInputFormat implements the factory method createRecordReader().

The following is an excerpt of the source code for FileInputFormat taken as is from the hadoop-1.2.1:

Listing-3

/**
 * 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.hadoop.mapreduce.lib.input;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.security.TokenCache;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.util.StringUtils;

/** 
 * A base class for file-based {@link InputFormat}s.
 * 
 * <p><code>FileInputFormat</code> is the base class for all file-based 
 * <code>InputFormat</code>s. This provides a generic implementation of
 * {@link #getSplits(JobContext)}.
 * Subclasses of <code>FileInputFormat</code> can also override the 
 * {@link #isSplitable(JobContext, Path)} method to ensure input-files are
 * not split-up and are processed as a whole by {@link Mapper}s.
 */
public abstract class FileInputFormat<K, V> extends InputFormat<K, V> {
  .
  .
  [--- Code here deleted for brevity ---]
  .
  .

  /** 
   * Generate the list of files and make them into FileSplits.
   */ 
  public List<InputSplit> getSplits(JobContext job
                                    ) throws IOException {
    long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
    long maxSize = getMaxSplitSize(job);

    // generate splits
    List<InputSplit> splits = new ArrayList<InputSplit>();
    List<FileStatus>files = listStatus(job);
    for (FileStatus file: files) {
      Path path = file.getPath();
      FileSystem fs = path.getFileSystem(job.getConfiguration());
      long length = file.getLen();
      BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);
      if ((length != 0) && isSplitable(job, path)) { 
        long blockSize = file.getBlockSize();
        long splitSize = computeSplitSize(blockSize, minSize, maxSize);

        long bytesRemaining = length;
        while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
          int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
          splits.add(new FileSplit(path, length-bytesRemaining, splitSize, 
                                   blkLocations[blkIndex].getHosts()));
          bytesRemaining -= splitSize;
        }
        
        if (bytesRemaining != 0) {
          splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining, 
                     blkLocations[blkLocations.length-1].getHosts()));
        }
      } else if (length != 0) {
        splits.add(new FileSplit(path, 0, length, blkLocations[0].getHosts()));
      } else { 
        //Create empty hosts array for zero length files
        splits.add(new FileSplit(path, 0, length, new String[0]));
      }
    }
    
    // Save the number of input files in the job-conf
    job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());

    LOG.debug("Total # of splits: " + splits.size());
    return splits;
  }
  
  .
  .
  [--- Code here deleted for brevity ---]
  .
  .
}

From the Listing-3 above, in the method getSplits(), the logic is to go through each of the input file(s) and return a list of FileSplits.

Notice that the method getSplits() references the class org.apache.hadoop.fs.FileSystem to determine the splits.

For HDFS, the FileSystem used is an instance of org.apache.hadoop.hdfs.DistributedFileSystem.

Now that we have peeked into the functionality of some of the important classes above, it is time to look at the TaskTracker.

Here is what the TaskTracker does:

Fetches all the resources (Java jar, the configuration information, and the allocated InputSplit from HDFS
Periodically sends a heartbeat to the JobTracker to report status on the assigned task
On start of the Mapper task, creates a local temporary directory to store the output from the Mapper task
For the Mapper task, uses RecordReader to fetch the next record, and invokes the map() function with the input record as the input value
The map() function outputs a KEY-VALUE pair as output, which is run through the org.apache.hadoop.mapreduce.Partitioner so that output for the same KEY are combined into a set of KEY-LIST<VALUE> pairs. The KEY-LIST<VALUE> pairs are then sorted by the KEY
When the Mapper task completes, all the sorted KEY-LIST<VALUE> pairs are saved to a file as output in the local temporary directory and the JobTracker is notified of the completion
For the Reducer task, it fetches the output file (containing the sorted KEY-LIST<VALUE> pairs) from the nodes that executed the Mapper tasks and invokes the reduce() function for each KEY-LIST<VALUE> pair
The Reducer task uses an implementation of org.apache.hadoop.mapreduce.OutputFormat to write to the output directory in HDFS

In the above paragraph describing the TaskTracker, we indicated few classes such as the RecordReader, Partitioner, OutputFormat, etc. Let us elaborate on them in the following paragraph(s).

RecordReader

Used by the Mapper task in the TaskTracker
An InputSplit provides a raw byte-oriented view of the input data block, while the org.apache.hadoop.mapreduce.RecordReader provides a record-oriented view of the input data
Provides input value into the map() function of the Mapper
The default implementation for RecordReader is org.apache.hadoop.mapreduce.lib.input.LineRecordReader
The LineRecordReader is what provides each line (terminated by a newline) as a record to the Mapper

The following is the source code for RecordReader taken as is from the hadoop-1.2.1:

Listing-4

/**
 * 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.hadoop.mapreduce;

import java.io.Closeable;
import java.io.IOException;

/**
 * The record reader breaks the data into key/value pairs for input to the
 * {@link Mapper}.
 * @param <KEYIN>
 * @param <VALUEIN>
 */
public abstract class RecordReader<KEYIN, VALUEIN> implements Closeable {

  /**
   * Called once at initialization.
   * @param split the split that defines the range of records to read
   * @param context the information about the task
   * @throws IOException
   * @throws InterruptedException
   */
  public abstract void initialize(InputSplit split,
                                  TaskAttemptContext context
                                  ) throws IOException, InterruptedException;

  /**
   * Read the next key, value pair.
   * @return true if a key/value pair was read
   * @throws IOException
   * @throws InterruptedException
   */
  public abstract 
  boolean nextKeyValue() throws IOException, InterruptedException;

  /**
   * Get the current key
   * @return the current key or null if there is no current key
   * @throws IOException
   * @throws InterruptedException
   */
  public abstract
  KEYIN getCurrentKey() throws IOException, InterruptedException;
  
  /**
   * Get the current value.
   * @return the object that was read
   * @throws IOException
   * @throws InterruptedException
   */
  public abstract 
  VALUEIN getCurrentValue() throws IOException, InterruptedException;
  
  /**
   * The current progress of the record reader through its data.
   * @return a number between 0.0 and 1.0 that is the fraction of the data read
   * @throws IOException
   * @throws InterruptedException
   */
  public abstract float getProgress() throws IOException, InterruptedException;
  
  /**
   * Close the record reader.
   */
  public abstract void close() throws IOException;
}

As can be inferred from Listing-4 above, RecordReader is an abstract class with no implementation.

The concrete implementation for RecordReader for reading text input files is the class LineRecordReader.

The following is the source code for LineRecordReader taken as is from the hadoop-1.2.1:

Listing-5

/**
 * 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.hadoop.mapreduce.lib.input;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.Seekable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.CodecPool;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.Decompressor;
import org.apache.hadoop.io.compress.SplitCompressionInputStream;
import org.apache.hadoop.io.compress.SplittableCompressionCodec;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.util.LineReader;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.logging.Log;

/**
 * Treats keys as offset in file and value as line. 
 */
public class LineRecordReader extends RecordReader<LongWritable, Text> {
  private static final Log LOG = LogFactory.getLog(LineRecordReader.class);

  private CompressionCodecFactory compressionCodecs = null;
  private long start;
  private long pos;
  private long end;
  private LineReader in;
  private int maxLineLength;
  private LongWritable key = null;
  private Text value = null;
  private Seekable filePosition;
  private CompressionCodec codec;
  private Decompressor decompressor;

  public void initialize(InputSplit genericSplit,
                         TaskAttemptContext context) throws IOException {
    FileSplit split = (FileSplit) genericSplit;
    Configuration job = context.getConfiguration();
    this.maxLineLength = job.getInt("mapred.linerecordreader.maxlength",
                                    Integer.MAX_VALUE);
    start = split.getStart();
    end = start + split.getLength();
    final Path file = split.getPath();
    compressionCodecs = new CompressionCodecFactory(job);
    codec = compressionCodecs.getCodec(file);

    // open the file and seek to the start of the split
    FileSystem fs = file.getFileSystem(job);
    FSDataInputStream fileIn = fs.open(split.getPath());

    if (isCompressedInput()) {
      decompressor = CodecPool.getDecompressor(codec);
      if (codec instanceof SplittableCompressionCodec) {
        final SplitCompressionInputStream cIn =
          ((SplittableCompressionCodec)codec).createInputStream(
            fileIn, decompressor, start, end,
            SplittableCompressionCodec.READ_MODE.BYBLOCK);
        in = new LineReader(cIn, job);
        start = cIn.getAdjustedStart();
        end = cIn.getAdjustedEnd();
        filePosition = cIn;
      } else {
        in = new LineReader(codec.createInputStream(fileIn, decompressor),
            job);
        filePosition = fileIn;
      }
    } else {
      fileIn.seek(start);
      in = new LineReader(fileIn, job);
      filePosition = fileIn;
    }
    // If this is not the first split, we always throw away first record
    // because we always (except the last split) read one extra line in
    // next() method.
    if (start != 0) {
      start += in.readLine(new Text(), 0, maxBytesToConsume(start));
    }
    this.pos = start;
  }
  
  private boolean isCompressedInput() {
    return (codec != null);
  }

  private int maxBytesToConsume(long pos) {
    return isCompressedInput()
      ? Integer.MAX_VALUE
      : (int) Math.min(Integer.MAX_VALUE, end - pos);
  }

  private long getFilePosition() throws IOException {
    long retVal;
    if (isCompressedInput() && null != filePosition) {
      retVal = filePosition.getPos();
    } else {
      retVal = pos;
    }
    return retVal;
  }

  public boolean nextKeyValue() throws IOException {
    if (key == null) {
      key = new LongWritable();
    }
    key.set(pos);
    if (value == null) {
      value = new Text();
    }
    int newSize = 0;
    // We always read one extra line, which lies outside the upper
    // split limit i.e. (end - 1)
    while (getFilePosition() <= end) {
      newSize = in.readLine(value, maxLineLength,
          Math.max(maxBytesToConsume(pos), maxLineLength));
      if (newSize == 0) {
        break;
      }
      pos += newSize;
      if (newSize < maxLineLength) {
        break;
      }

      // line too long. try again
      LOG.info("Skipped line of size " + newSize + " at pos " + 
               (pos - newSize));
    }
    if (newSize == 0) {
      key = null;
      value = null;
      return false;
    } else {
      return true;
    }
  }

  @Override
  public LongWritable getCurrentKey() {
    return key;
  }

  @Override
  public Text getCurrentValue() {
    return value;
  }

  /**
   * Get the progress within the split
   */
  public float getProgress() throws IOException {
    if (start == end) {
      return 0.0f;
    } else {
      return Math.min(1.0f,
        (getFilePosition() - start) / (float)(end - start));
    }
  }

  public synchronized void close() throws IOException {
    try {
      if (in != null) {
        in.close();
      }
    } finally {
      if (decompressor != null) {
        CodecPool.returnDecompressor(decompressor);
      }
    }
  }
}

From the Listing-5 above, in the method initialize(), one of the steps is to open the data input stream for the specified InputSplit. In otherwords, it references the class org.apache.hadoop.fs.FSDataInputStream for the data input stream.

For HDFS, the FSDataInputStream used is an instance of org.apache.hadoop.hdfs.DFSClient$DFSDataInputStream.

DFSClient$DFSDataInputStream internally uses an instance of org.apache.hadoop.hdfs.DFSClient$DFSInputStream.

It is DFSClient$DFSInputStream that handles all the communication with the NameNode and various DataNodes for the various splits (data blocks) of an input file.

Partitioner

Used in the TaskTracker after the Mapper task produces output
Helps in the distribution of output from the Mapper to the Reducer
The default implementation for Partitioner used is org.apache.hadoop.mapred.lib.HashPartitioner. It uses the hash code on KEY to distribute the output

OutputFormat

Used by the Reducer task in the TaskTracker
Used for writing output to HDFS from the Reducer task
The default implementation for OutputFormat is org.apache.hadoop.mapreduce.lib.input.TextOutputFormat
Is a factory for org.apache.hadoop.mapreduce.RecordWriter

RecordWriter

Used by the Reducer task in the TaskTracker
The default implementation for RecordWriter is org.apache.hadoop.mapreduce.lib.input.TextOutputFormat.LineRecordWriter
The TextOutputFormat.LineRecordWriter is what outputs each line of KEY-VALUE pairs (terminated by a newline) as a record from the Reducer

The following Figure-2 illustrates the end-to-end picture of the classes:

Figure-2

Note that the sort step is really not a class and hence depicted as a box with dotted boundary.