Situation: In this first example, the input data is several million lines of plain-text phone call records. Each CSV input line is of the format:

{date},{caller num},{caller carrier},{dest num},{dest carrier},{length}

The plan is to analyze each call record separately, so there’s no need to sort and group the data. In turn, we won’t need the full MapReduce cycle but can use a Map-only job to distribute the work throughout the cluster.

The code: To analyze each call record, consider a function callAnalysis() that takes all of the record’s fields as parameters:

callAnalysis( date , caller.num, caller.carrier , dest.num , dest.carrier , length )

Hadoop streaming does not invoke R functions directly. You provide an R script that calls the functions, and Hadoop invokes your R script. Specifically, Hadoop will pass an entire input record to the Map operation R script via standard input. It’s up to your R script to disassemble the record into its components (here, split it by commas) and feed it into the function (see Example 6-1).

#! /usr/bin/env Rscript
input <- file( "stdin" , "r" )
while( TRUE ){
   currentLine <- readLines( input , n=1 )  

   if( 0 == length( currentLine ) ){
       break
   }

   currentFields <- unlist( strsplit( currentLine , "," ) ) 

   result <- callAnalysis(
       currentFields[1] , currentFields[2] , currentFields[3] ,
       currentFields[4] , currentFields[5] , currentFields[6]
   ) 

   cat( result , "
" , sep="" ) 
}
close( input )

This may not be the most efficient code. That’s alright. Large-scale parallelism tends to wash away smaller code inefficiencies.

Put another way, clustered computer power is cheap compared to human thinking-power. Save your brain for solving data-related problems and let Hadoop pick up any slack. Your R code would have to be extremely inefficient before an extensive tuning exercise would yield a payoff.

cat input-sample.txt | ./mapper.R | sort | ./reducer.R

Running the Hadoop job:

export HADOOP_VERSION="0.20.203.0"
export HADOOP_HOME="/opt/thirdparty/dist/hadoop-${HADOOP_VERSION}"
export HADOOP_COMMAND="${HADOOP_HOME}/bin/hadoop"
export HADOOP_STREAMING_JAR="${HADOOP_HOME}/contrib/streaming/hadoop-streaming-
${HADOOP_VERSION}.jar"
export HADOOP_COMPRESSION_CODEC="org.apache.hadoop.io.compress.GzipCodec"
export HADOOP_INPUTFORMAT="org.apache.hadoop.mapred.lib.NLineInputFormat"

${HADOOP_COMMAND} jar ${HADOOP_STREAMING_JAR} 
 
 -D mapreduce.job.reduces=0  
 -D mapred.output.compress=true  
 -D mapred.output.compression.codec=${HADOOP_COMPRESSION_CODEC} 
 -D mapred.task.timeout=600000  
 
 -inputformat ${HADOOP_INPUTFORMAT}  
 -input /tmp/call-records.csv 
 -output /tmp/hadoop-out 
 -mapper $PWD/mapper.R

This is a pretty standard command line for Hadoop streaming: it specifies the streaming JAR, the mapper script, and the input and output locations. Note that the “generic” Hadoop options, which begin with -D, always come first.

There are a few points of note:

(This code works for Hadoop 0.19 and 0.20. For Hadoop 0.21, use org.apache.hadoop.mapreduce.lib.input.NLineInputFormat. Note the subtle difference in class name!)

Reviewing the output: A typical Hadoop job will create several files, one for each Reduce operation. Since this is a Map-only job, there is just one file, called /tmp/hadoop-out/part-0000.gz. You can use gunzip to uncompress the file and then review the contents.

Anything the script wrote to standard output would be included this output file. That is, Hadoop doesn’t discern between the lines you want in the output (such as the output from the fictional callAnalysis() function) and any lingering cat() or print() calls in your code or in the modules you load. If you find stray content in your job’s output, you can post-process those files or suppress the output in your code. Here, “post-process” is a fancy term for grep’ing the job output file to extract the lines of interest. “Suppress” means you call sink() to selectively disable standard output:

sink( "/dev/null" ) ## suppress standard output

... do the work ...

sink() ## restore standard output

cat(  ... your intended result ... )
    ... exit the script

As of this writing, Hadoop Streaming can only be used for text input and output. (This is slated to change in a future Hadoop release.) This doesn’t preclude you from working with binary data in a streaming job; but it does preclude your Map and Reduce scripts from accepting binary input and producing binary output. This example presents a workaround.

Situation: Imagine that you want to analyze a series of image files. Perhaps they are frames from a video recording, or a file full of serialized R objects, or maybe you run a large photo-sharing site. For this example, let’s say you have R code that will perform image feature extraction. Each image is an independent unit of work, so assuming you have a decent number of images (and/or feature extraction takes some noticeable amount of time), a Map-only Hadoop job is a good fit.

The code: Remember, though, that Hadoop Streaming can only handle line-by-line text input and output so you have to get creative here. One option would be to feed your Hadoop Streaming job an input of pointers to the data, which your R script could then fetch and process locally. For example:

(Notice that I don’t mention using HDFS. Remember that HDFS doesn’t work well with small files. I’ll cover a different approach in the next example.)

For this example, let’s say the mythical imageFeatureExtraction() function accepts an R url() connection:

#! /usr/bin/env Rscript
input <- file( "stdin" , "r" )
while( TRUE ){
   currentLine <- readLines( input , n=1 )

   if( 0 == length( currentLine ) ){
       break
   }

   pulledData <- url( currentLine ) )

   result <- imageFeatureExtraction( url( currentLine ) )
   cat( result , "
" , sep="" )
}
close( input )

Running the Hadoop job: As far as Hadoop is concerned, this example looks just like the previous one (all the changes are in the R script), so it uses the same command line.

Reviewing the output: If your job outputs text (such as a series of numbers describing the binary data) then you can send that to standard output, as usual.

Let’s say your job yields binary output, such as a series of charts. In that case, you can use the same idea as you did for the input, and push the output to another system:

and so on.

Caveats: Keep in mind that this method is not perfect.

A Hadoop cluster is a robust beast. Between the software-side framework and the required hardware layout, you are protected from hard disk failures, node crashes, and even loss of network connectivity.

To reap this benefit, though, the job must be self-contained from Hadoop’s perspective. Everything required for the job must exist within the cluster. Map and Reduce scripts, input data, and output must all live in HDFS (S3 if you are using Elastic MapReduce).

Once you rely on systems or services outside of the cluster, you lose in four ways:

This may sound like a long list of caveats, but whether it’s really a hurdle depends on your circumstances. Leveraging external services and side effects will work just fine if most of the “action” will take place inside Hadoop, or the external data is just a small part of the process, or multiple execution is not a problem.

That said, there is another way to handle binary data in a streaming job but still keep all the data in the cluster. It’s slightly more involved than the methods described in this example, but the trade-off may be worth your while. In the next section, I’ll show you how to do this using the Java API instead of Streaming.

Situation: You still wish to perform feature extraction on a series of image files, but you feel the previous solution is too fragile. You want the job to be self-contained, from a Hadoop perspective.

The code: Recall, from the previous chapter, that if you’re processing whole-file data (text or binary) with Hadoop, you’ll need to pack those files into a SequenceFile archive. The streaming API cannot presently handle SequenceFiles but the Java API can. You can use the Java API to extract data from the SequenceFile input, write it to a predictable filename, and launch an R script that operates on that file.

Once again, we need Hadoop for a one-stage parallel execution, so this will be a Map-only job.

First, let’s look at the Java code in Example 6-2. Hadoop Java code is typically compiled into a JAR that has at least two classes: the driver that configures the job and the mapper that is run for a Map task. (Jobs that have a Reduce step will include a class for the Reduce task.)

public class Driver extends Configured implements Tool { 

  public int run( String[] args ) throws Exception {
     final Job job = new Job( getConf() ) ;

     job.setJobName( "Parallel R: Chapter 5, example 3" ) ;
     job.setMapperClass( Ch5Ex3Mapper.class ) ; 

     FileInputFormat.addInputPath( job , new Path( args[0] ) ) ; 
     FileOutputFormat.setOutputPath( job , new Path( args[1] ) ) ;
     job.setInputFormatClass( SequenceFileInputFormat.class ) ; 

     job.setOutputKeyClass( Text.class ) ; 
     job.setOutputValueClass( Text.class ) ;

     job.getConfiguration().setBoolean( "mapred.output.compress" , true ) ; 

     job.getConfiguration().setClass( "mapred.output.compression.codec" ,
       BZip2Codec.class , CompressionCodec.class ) ;
     job.getConfiguration().setInt( "mapreduce.job.reduces" , 0 ) ; 

     return( job.waitForCompletion( true ) ? 0 : 1 ) ;

  }
}

public class Ch5Ex3Mapper extends Mapper<Text, BytesWritable , Text , Text> { 

private final Text _outputValue ;  
private final StringBuilder _rOutputBuffer ;

public void map(Text inputKey , BytesWritable inputValue , Context context )
   throws IOException , InterruptedException {

     _outputValue.clear() ;  
     if( _rOutputBuffer.length() > 0 ){
         _rOutputBuffer.delete( 0 , _rOutputBuffer.length() ) ;
     }

     BufferedReader rOutputReader = null ;
     OutputStream fileWriteHandle = null ;
     final File currentFile = new File( inputKey.toString() ) ;

     try{

         // write the raw bytes to a file. (input key name is the file name)
         fileWriteHandle = new FileOutputStream( currentFile ) ;  
         fileWriteHandle.write( inputValue.getBytes() , 0, inputValue.getLength() ) ;
         closeOutputStream( fileWriteHandle ) ;

         final ArrayList<String> tempList = new ArrayList<String>() ;  
         final List<String> commandLine = new ArrayList<String>() ;
         commandLine.add( "/usr/bin/env" ) ;
         commandLine.add( "R" ) ;
         commandLine.add( "--vanilla" ) ;
         commandLine.add( "--slave" ) ;
         commandLine.add( "--file=helper.R" ) ;
         commandLine.add( "--args " ) ;
         commandLine.add( inputKey.toString() ) ;

         final Process runtime = new ProcessBuilder( commandLine )
             .redirectErrorStream( true )
             .start() ; 

         final int exitCode = runtime.waitFor() ;
         rOutputReader = new BufferedReader(
           new InputStreamReader( runtime.getInputStream() ) ) ;

         if( 0 != exitCode ){
             _rOutputBuffer.append( "error! " ) ;
         }

         _rOutputBuffer.append( rOutputReader.readLine() ) ;  
         _outputValue.set( _rOutputBuffer.toString() ) ;

         context.write( inputKey ,  _outputValue ) ;  


     }catch( final Exception rethrow ){
         throw( new IOException( rethrow ) ) ;
     }finally{
             // … close handles and delete the image file ...
     }

  }

}

Now let’s look at the R script, helper.R, which is invoked by the Java code in Example 6-4:

dataFile <- commandArgs(trailingOnly=TRUE)  
result <- imageFeatureExtraction( dataFile )  
output.value <- paste( dataFile , result , sep="	" )

Running the Hadoop job: Let’s say the Hadoop code is in a JAR named “launch-R.jar” and the input images are in a SequenceFile named images-sample.seq. Assuming the environment variables defined above, you can launch the job as follows:

${HADOOP_COMMAND} jar launch-R.jar 
   -files helper.R  
   /tmp/images-sample.seq 
   /tmp/hadoop-out

This command line is much shorter than the streaming command lines, mostly because several options are set in the driver class.

This is a typical Hadoop command line for Java jobs, with one exception:

If you’re testing the job on your workstation, in Hadoop’s “local” (single-workstation) mode, you’ll want to keep two ideas in mind:

For one, Distributed Cache doesn’t really work in local mode. You’ll want to launch the Hadoop command from the directory where helper.R lives such that the Mapper class can find it.

Secondly, the Hadoop job will treat the current directory as its runtime directory. That means the images extracted in the Mapper will be written to your current directory.

Reviewing the output: Let’s assume the feature extraction function yields text output, so you would fetch and read it as explained in the previous examples. If your R code generates binary output, such as charts, you can write that as a SequenceFile: specify SequenceFileOutputFormat as the output, and have your Java code write the file’s data to a BytesWritable object.

Caveats: This method keeps the entire job within Hadoop’s walls: unlike the previous example, you’re protected from machine crashes, network failures, and Hadoop’s speculative execution. The cost is the extra overhead involved in putting all of your input data in SequenceFiles. Even if you’re using forqlift, or you’re comfortable writing a Hadoop Java job to do this, you still need to gather the inputs. That may involve a separate effort to copy the data from an existing service, such as an internal web server or file server.

Situation: You want to collect related records and operate on that group as a whole.

Returning to the “phone records” example, let’s say you want to analyze every number’s output call patterns. That would require you to first gather all of the calls made by each number (Map phase) and then process those records together (Reduce phase).

The code: As noted above, this will require both the Map and Reduce phases.

The Map phase code will extract the caller’s phone number to use as the key, as shown in Example 6-4.

#! /usr/bin/env Rscript

input <- file( "stdin" , "r" )
while( TRUE ){
   currentLine <- readLines( input , n=1 )
   if( 0 == length( currentLine ) ){
       break
   }

   currentFields <- unlist( strsplit( currentLine , "," ) )

   result <- paste(
       currentFields[1] ,
       currentLine ,
       sep="	"
   )

   cat( result , "
" )
}
close( input )

By now, the basic structure should look familiar. Remember that the first field in the comma-separated line is the caller’s phone number, which serves as the key output from the Map task.

The Reducer code builds a data.frame of all calls made by each number, then passes the data.frame to the analysis function. Something to note here is that the logic in a Reducer script is different from that of a Map script. The flow may seem a little strange, so I’ll explain it at a high level before showing the code sample.

In a Reducer script, each input line is of the format:

{key}{tab}{value}

where {key} and {value} are a single key/value pair, as output from a Map task.

Recall that the Reducer’s job is to collect all of the values for a given key, then process them together. Hadoop may pass a single Reducer values for multiple keys, but it will sort them first. When the key changes, then, you know you’ve seen all of the values for the previous key. You can process those values as a group, then move on to the next key.

With that in mind, a sample Reducer script for the call-analysis job is shown in Example 6-5.

#! /usr/bin/env Rscript

input <- file( "stdin" , "r" )
lastKey <- ""

tempFile <- tempfile( pattern="hadoop-mr-demo-" , fileext="csv" ) 
tempHandle <- file( tempFile , "w" )

while( TRUE ){

   currentLine <- readLines( input , n=1 )
   if( 0 == length( currentLine ) ){
       break
   }

   tuple <- unlist( strsplit( currentLine , "	" ) ) 
   currentKey <- tuple[1]
   currentValue <- tuple[2]

   if( ( currentKey != lastKey ) ){
       if( lastKey != "" ){  
         close( tempHandle )
         bucket <- read.csv( tempFile , header=FALSE )  
         result <- anotherCallAnalysisFunction( bucket )  
         cat( currentKey , "	" , result , "
" )  
         tempHandle <- file( tempFile , "w" )  
       }

       lastKey <- currentKey  
   }

   cat( currentLine , "
" , file=tempHandle )  
}

close( tempHandle ) 

bucket <- read.csv( tempFile , header=FALSE )
result <- anotherCallAnalysisFunction( bucket )
cat( currentKey , "	" , result , "
" )

unlink( tempFile )

close( input )

Running the Hadoop job:

${HADOOP_COMMAND} jar ${HADOOP_STREAMING_JAR} 
   
   -D mapred.output.compress=true 
   -D mapred.output.compression.codec=${HADOOP_COMPRESSION_CODEC} 
   
   -inputformat ${HADOOP_INPUTFORMAT} 
   -input /tmp/call-records.csv 
   -output /tmp/hadoop-out 
   -mappermapper.R 
   -reducer reducer.R

This is a standard Hadoop streaming command line. The only deviation from previous streaming command lines is that this one specifies a reducer script.

Reviewing the output: By now, you know how to explore both text and binary job output. This section is intentionally left blank.

Caveats: When you use the full Map and Reduce phases, you need to know how your data is distributed in terms of the keys output from the Map phase. This may require you to do some up-front exploratory data analysis before you can determine whether the job is truly Hadoop-worthy.

Generally speaking, the Map phase is very lightweight (since it’s just used to assign keys to each input) and the heavy lifting takes place in the Reduce operation. To take advantage of the parallelism of the Reduce stage, then, you’ll need to meet two conditions:

Why is this? Say you have ten million input records. If the Map operation yields only two unique keys, each with five million records, then you will have two very long-running Reduce tasks and that would not be a scalable operation.

Alternatively, let’s say the Map phase yields ten-thousand unique keys, but one of those keys has several million records. This would yield an unbalanced Reduce phase, in which the work for one key takes so long that it eliminates the gains from running the remaining keys’ work in parallel.

I expect the phone records example is still a good fit for Hadoop parallelization since it is highly unlikely that one phone number made most of the calls. For your jobs, though, this may not be such a safe assumption.

This chapter covered a lot of ground, so let’s take a step back to review when you’d want to use R+Hadoop and when you’d want to try another method.