Is Hadoop MapReduce MapContext.write() thread safe?

user_421049 0
Viewed 158

Traditionally a Hadoop MapReduce mapper sequentially processes data and emits values by writing to the map context, as this example excerpt from the org.apache.hadoop.mapreduce.Mapper API docs shows:

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

But what if the mapper decides to start up several threads, each of them doing a portion of the work (sort of a further divide-and-conquer within the mapper)? Each one will want to emit data when it is ready.

Does MapContext.write() (really TaskInputOutputContext.write(KEYOUT key, VALUEOUT value)) support being called concurrently within the mapper? Or must I synchronize the calls to MapContext.write() and guarantee they are called sequentially? (And what sort of bad things would happen if I should and I don't?)

(I anticipate replies saying I shouldn't be starting up multiple threads in a mapper in the first place. While I respect that point of view, pragmatically this is what I'm working with at the moment.)

hadoop java
1 Answers

No, it isn't by default. That's why they let you override the run() method, which handles the flow of the mapper operations.

This is the source code, and default implementation for run:

public void run(Context context) throws IOException, InterruptedException {
    setup(context);
    while (context.nextKeyValue()) {
      map(context.getCurrentKey(), context.getCurrentValue(), context);
    }
    cleanup(context);
}

And at the very end of the Mapper javaDoc, this phrase gives some light:

Applications may override the run(org.apache.hadoop.mapreduce.Mapper.Context) method to exert greater control on map processing e.g. multi-threaded Mappers etc.

So the idea is clear; You want thread-safety on multithread environment, that's cool...but you implement the logic. A very basic example of this could be (I synch all, because I don't know the details of each method/element, just for the example):

@Override
public void run(Context context) throws IOException, InterruptedException 
{
    synchronized(lock)   //shared lock, class lock, etc..
   { 
      setup(context);
      while (context.nextKeyValue()) 
        map(context.getCurrentKey(), context.getCurrentValue(), context);
      
      cleanup(context);
    }
}

Although there's no specific mention to multithreading with this method, something similar happens with map():

protected void map

Called once for each key/value pair in the input split. Most applications should override this, but the default is the identity function.

In resume, overriding these methods (at least one of them), should let you create a multithread Mapper extension.

So yes, it can be thread-safe, but it's your job to do so. The default implementation is not.

Update - MultithreadedMapper

Well, they do offer a multithreaded runnable mapper implementation: Anyway, seems like the synchronization regarding the map/write operations are still your job; This just implements the mechanism to invoke run() from different pooled threads.

MultithreadedMapper

Multithreaded implementation for @link org.apache.hadoop.mapreduce.Mapper. It can be used instead of the default implementation, MapRunner, when the Map operation is not CPU bound in order to improve throughput.

Mapper implementations using this MapRunnable must be thread-safe.

The Map-Reduce job has to be configured with the mapper to use via setMapperClass(Job, Class) and the number of thread the thread-pool can use with the getNumberOfThreads(JobContext) method. The default value is 10 threads.

The key here seems to be the setMapperClass() method; There a custom thread-safe extension of Mapper should be passed as argument.

-- Appendix - MultithreadedZipContentLoader

The interesting part is in main(), as defines the protocol to use the MultiMapper

public class MultithreadedZipContentLoader {
    public static class ZipContentMapper extends Mapper<Text, Text, DocumentURI, Text> 
     {
        private DocumentURI uri = new DocumentURI();
         
        public void map(Text fileName, Text fileContent, Context context) 
        throws IOException, InterruptedException 
        {
              uri.setUri(fileName.toString());
              context.write(uri, fileContent);
        }
    }
    
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length < 2) {
            System.err.println("Usage: MultithreadedZipContentLoader configFile inputDir threadCount");
            System.exit(2);
        }
        
        Job job = Job.getInstance(conf);
        job.setJarByClass(MultithreadedZipContentLoader.class);
        job.setInputFormatClass(ZipContentInputFormat.class);
        job.setMapperClass(MultithreadedMapper.class);
        MultithreadedMapper.setMapperClass(job, ZipContentMapper.class);
        MultithreadedMapper.setNumberOfThreads(job, Integer.parseInt(args[2]));
        job.setMapOutputKeyClass(DocumentURI.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputFormatClass(ContentOutputFormat.class);
        
        ZipContentInputFormat.setInputPaths(job, new Path(otherArgs[1]));

        conf = job.getConfiguration();
        conf.addResource(otherArgs[0]);
         
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

class ZipContentInputFormat extends FileInputFormat<Text, Text> {

    @Override
    protected boolean isSplitable(JobContext context, Path filename) {
        return false;
    }
    
    @Override
    public RecordReader<Text, Text> createRecordReader(InputSplit split,
            TaskAttemptContext context) throws IOException, InterruptedException {
        return new ZipContentReader();
    }
    
}

...

The only thing that this example lacks is the synchronize part in the extended map method, as in this case seems there's no need to (no duplicate file uris, I guess).

user_21489530
Related