org.infinispan.distexec.mapreduce

Class MapReduceTask<KIn,VIn,KOut,VOut>

java.lang.Object

org.infinispan.distexec.mapreduce.MapReduceTask<KIn,VIn,KOut,VOut>

public class MapReduceTask<KIn,VIn,KOut,VOut>
extends java.lang.Object

MapReduceTask is a distributed task allowing a large scale computation to be transparently parallelized across Infinispan cluster nodes.

Users should instantiate MapReduceTask with a reference to a cache whose data is used as input for this task. Infinispan execution environment will migrate and execute instances of provided Mapper and Reducer seamlessly across Infinispan nodes.

Unless otherwise specified using onKeys(Object...) filter all available key/value pairs of a specified cache will be used as input data for this task. For example, MapReduceTask that counts number of word occurrences in a particular cache where keys and values are String instances could be written as follows:

 MapReduceTask<String, String, String, Integer> task = new MapReduceTask<String, String, String, Integer>(cache);
 task.mappedWith(new WordCountMapper()).reducedWith(new WordCountReducer());
 Map<String, Integer> results = task.execute();
 

The final result is a map where key is a word and value is a word count for that particular word.

Accompanying Mapper and Reducer are defined as follows:

    private static class WordCountMapper implements Mapper<String, String, String,Integer> {

     public void map(String key, String value, Collector<String, Integer> collector) {
        StringTokenizer tokens = new StringTokenizer(value);
       while (tokens.hasMoreElements()) {
           String s = (String) tokens.nextElement();
           collector.emit(s, 1);
        }
     }
  }

   private static class WordCountReducer implements Reducer<String, Integer> {

      public Integer reduce(String key, Iterator<Integer> iter) {
         int sum = 0;
         while (iter.hasNext()) {
            Integer i = (Integer) iter.next();
            sum += i;
        }
         return sum;
      }
   }
 

Finally, as of Infinispan 5.2 release, MapReduceTask can also specify a Combiner function. The Combiner is executed on each node after the Mapper and before the global reduce phase. The Combiner receives input from the Mapper's output and the output from the Combiner is then sent to the reducers. It is useful to think of the Combiner as a node local reduce phase before global reduce phase is executed.

Combiners are especially useful when reduce function is both commutative and associative! In such cases we can use the Reducer itself as the Combiner; all one needs to do is to specify the Combiner:

 MapReduceTask<String, String, String, Integer> task = new MapReduceTask<String, String, String, Integer>(cache);
 task.mappedWith(new WordCountMapper()).reducedWith(new WordCountReducer()).combineWith(new WordCountReducer());
 Map<String, Integer> results = task.execute();
 

Note that Mapper and Reducer should not be specified as inner classes. Inner classes declared in non-static contexts contain implicit non-transient references to enclosing class instances, serializing such an inner class instance will result in serialization of its associated outer class instance as well.

If you are not familiar with concept of map reduce distributed execution model start with Google's MapReduce research paper.

Since:

5.0

Author:

Manik Surtani, Vladimir Blagojevic, Sanne Grinovero

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
private static interface	MapReduceTask.CancellableTaskPart
private class	MapReduceTask.MapReduceTaskFuture<R>
private class	MapReduceTask.MapTaskPart<V>
private class	MapReduceTask.ReduceTaskPart<V>
private class	MapReduceTask.TaskPart<V>

Field Summary

Fields

Modifier and Type	Field and Description
protected AdvancedCache<KIn,VIn>	cache
protected java.util.List<MapReduceTask.CancellableTaskPart>	cancellableTasks
protected org.infinispan.commands.CancellationService	cancellationService
protected org.infinispan.interceptors.locking.ClusteringDependentLogic	clusteringDependentLogic
protected Reducer<KOut,VOut>	combiner
static java.lang.String	DEFAULT_TMP_CACHE_CONFIGURATION_NAME
protected boolean	distributeReducePhase
protected boolean	isLocalOnly
protected java.util.Collection<KIn>	keys
private static org.infinispan.util.logging.Log	log
protected Mapper<KIn,VIn,KOut,VOut>	mapper
protected MapReduceManager	mapReduceManager
protected Marshaller	marshaller
protected Reducer<KOut,VOut>	reducer
protected org.infinispan.remoting.rpc.RpcOptionsBuilder	rpcOptionsBuilder
protected java.util.UUID	taskId
protected boolean	useIntermediateSharedCache

Constructor Summary

Constructors

Constructor and Description
MapReduceTask(Cache<KIn,VIn> masterCacheNode) Create a new MapReduceTask given a master cache node.
MapReduceTask(Cache<KIn,VIn> masterCacheNode, boolean distributeReducePhase) Create a new MapReduceTask given a master cache node.
MapReduceTask(Cache<KIn,VIn> masterCacheNode, boolean distributeReducePhase, boolean useIntermediateSharedCache) Create a new MapReduceTask given a master cache node.

Method Summary

Methods

Modifier and Type	Method and Description
protected void	aggregateReducedResult(java.util.Map<KOut,java.util.List<VOut>> finalReduced, java.util.Map<KOut,VOut> mapReceived)
private org.infinispan.commands.CancelCommand	buildCancelCommand(MapReduceTask.CancellableTaskPart taskPart)
private org.infinispan.commands.read.MapCombineCommand<KIn,VIn,KOut,VOut>	buildMapCombineCommand(java.lang.String taskId, Mapper<KIn,VIn,KOut,VOut> m, Reducer<KOut,VOut> r, java.util.Collection<KIn> keys, boolean reducePhaseDistributed, boolean emitCompositeIntermediateKeys)
private org.infinispan.commands.read.ReduceCommand<KOut,VOut>	buildReduceCommand(java.lang.String taskId, java.lang.String destinationCache, Reducer<KOut,VOut> r, java.util.Collection<KOut> keys, boolean emitCompositeIntermediateKeys)
protected Mapper<KIn,VIn,KOut,VOut>	clone(Mapper<KIn,VIn,KOut,VOut> mapper)
protected Reducer<KOut,VOut>	clone(Reducer<KOut,VOut> reducer)
MapReduceTask<KIn,VIn,KOut,VOut>	combinedWith(Reducer<KOut,VOut> combiner) Specifies Combiner to use for this MapReduceTask
protected <V> MapReduceTask.ReduceTaskPart<V>	createReducePart(org.infinispan.commands.read.ReduceCommand<KOut,VOut> cmd, org.infinispan.remoting.transport.Address target, java.lang.String destCacheName)
protected <V> MapReduceTask.MapTaskPart<V>	createTaskMapPart(org.infinispan.commands.read.MapCombineCommand<KIn,VIn,KOut,VOut> cmd, org.infinispan.remoting.transport.Address target, boolean distributedReduce)
protected boolean	distributeReducePhase()
private void	ensureProperCacheState(AdvancedCache<KIn,VIn> cache)
boolean	equals(java.lang.Object obj)
java.util.Map<KOut,VOut>	execute() Executes this task across Infinispan cluster nodes.
<R> R	execute(Collator<KOut,VOut,R> collator) Executes this task across Infinispan cluster but the final result is collated using specified Collator
java.util.concurrent.Future<java.util.Map<KOut,VOut>>	executeAsynchronously() Executes this task across Infinispan cluster nodes asynchronously.
<R> java.util.concurrent.Future<R>	executeAsynchronously(Collator<KOut,VOut,R> collator) Executes this task asynchronously across Infinispan cluster; final result is collated using specified Collator and wrapped by Future
protected java.util.Set<KOut>	executeMapPhase(boolean useCompositeKeys)
protected java.util.Map<KOut,VOut>	executeMapPhaseWithLocalReduction()
protected java.util.Map<KOut,VOut>	executeReducePhase(java.util.Set<KOut> allMapPhasesResponses, boolean useCompositeKeys)
protected void	executeTaskInit(java.lang.String tmpCacheName)
int	hashCode()
protected boolean	inputTaskKeysEmpty()
protected <T> java.util.Map<org.infinispan.remoting.transport.Address,? extends java.util.Collection<T>>	mapKeysToNodes(java.util.Collection<T> keysToMap)
protected <T> java.util.Map<org.infinispan.remoting.transport.Address,? extends java.util.Collection<T>>	mapKeysToNodes(java.util.Collection<T> keysToMap, boolean useIntermediateCompositeKey)
protected <T> java.util.Map<org.infinispan.remoting.transport.Address,? extends java.util.Collection<T>>	mapKeysToNodes(org.infinispan.distribution.DistributionManager dm, java.util.Collection<T> keysToMap, boolean useIntermediateCompositeKey)
MapReduceTask<KIn,VIn,KOut,VOut>	mappedWith(Mapper<KIn,VIn,KOut,VOut> mapper) Specifies Mapper to use for this MapReduceTask
private <K,V> void	mergeResponse(java.util.Map<K,java.util.List<V>> result, java.util.Map<K,java.util.List<V>> m)
MapReduceTask<KIn,VIn,KOut,VOut>	onKeys(KIn... input) Rather than use all available keys as input onKeys allows users to specify a subset of keys as input to this task
MapReduceTask<KIn,VIn,KOut,VOut>	reducedWith(Reducer<KOut,VOut> reducer) Specifies Reducer to use for this MapReduceTask
MapReduceTask<KIn,VIn,KOut,VOut>	timeout(long timeout, java.util.concurrent.TimeUnit unit) See timeout(TimeUnit).
long	timeout(java.util.concurrent.TimeUnit outputTimeUnit)
java.lang.String	toString()
protected boolean	useIntermediatePerTaskCache()
protected boolean	useIntermediateSharedCache()

Methods inherited from class java.lang.Object

clone, finalize, getClass, notify, notifyAll, wait, wait, wait

Field Detail

log

private static final org.infinispan.util.logging.Log log

DEFAULT_TMP_CACHE_CONFIGURATION_NAME

public static final java.lang.String DEFAULT_TMP_CACHE_CONFIGURATION_NAME

See Also:

Constant Field Values

mapper

protected Mapper<KIn,VIn,KOut,VOut> mapper

reducer

protected Reducer<KOut,VOut> reducer

combiner

protected Reducer<KOut,VOut> combiner

distributeReducePhase

protected final boolean distributeReducePhase

useIntermediateSharedCache

protected final boolean useIntermediateSharedCache

keys

protected final java.util.Collection<KIn> keys

cache

protected final AdvancedCache<KIn,VIn> cache

marshaller

protected final Marshaller marshaller

mapReduceManager

protected final MapReduceManager mapReduceManager

cancellationService

protected final org.infinispan.commands.CancellationService cancellationService

cancellableTasks

protected final java.util.List<MapReduceTask.CancellableTaskPart> cancellableTasks

taskId

protected final java.util.UUID taskId

clusteringDependentLogic

protected final org.infinispan.interceptors.locking.ClusteringDependentLogic clusteringDependentLogic

isLocalOnly

protected final boolean isLocalOnly

rpcOptionsBuilder

protected org.infinispan.remoting.rpc.RpcOptionsBuilder rpcOptionsBuilder

Constructor Detail

MapReduceTask

public MapReduceTask(Cache<KIn,VIn> masterCacheNode)

Create a new MapReduceTask given a master cache node. All distributed task executions will be initiated from this cache node. This task will by default only use distributed map phase while reduction will be executed on task originating Infinispan node.

Large and data intensive tasks whose reduction phase would exceed working memory of one Infinispan node should use distributed reduce phase

Parameters:

masterCacheNode - cache node initiating map reduce task

MapReduceTask

public MapReduceTask(Cache<KIn,VIn> masterCacheNode,
             boolean distributeReducePhase)

Create a new MapReduceTask given a master cache node. All distributed task executions will be initiated from this cache node.

Parameters:

masterCacheNode - cache node initiating map reduce task

distributeReducePhase - if true this task will use distributed reduce phase execution

MapReduceTask

public MapReduceTask(Cache<KIn,VIn> masterCacheNode,
             boolean distributeReducePhase,
             boolean useIntermediateSharedCache)

Create a new MapReduceTask given a master cache node. All distributed task executions will be initiated from this cache node.

Parameters:

masterCacheNode - cache node initiating map reduce task

distributeReducePhase - if true this task will use distributed reduce phase execution

useIntermediateSharedCache - if true this tasks will share intermediate value cache with other executing MapReduceTasks on the grid. Otherwise, if false, this task will use its own dedicated cache for intermediate values

Method Detail

onKeys

public MapReduceTask<KIn,VIn,KOut,VOut> onKeys(KIn... input)

Rather than use all available keys as input onKeys allows users to specify a subset of keys as input to this task

Parameters:

input - input keys for this task

Returns:

this task

mappedWith

public MapReduceTask<KIn,VIn,KOut,VOut> mappedWith(Mapper<KIn,VIn,KOut,VOut> mapper)

Specifies Mapper to use for this MapReduceTask

Note that Mapper should not be specified as inner class. Inner classes declared in non-static contexts contain implicit non-transient references to enclosing class instances, serializing such an inner class instance will result in serialization of its associated outer class instance as well.

Parameters:

mapper - used to execute map phase of MapReduceTask

Returns:

this MapReduceTask itself

reducedWith

public MapReduceTask<KIn,VIn,KOut,VOut> reducedWith(Reducer<KOut,VOut> reducer)

Specifies Reducer to use for this MapReduceTask

Note that Reducer should not be specified as inner class. Inner classes declared in non-static contexts contain implicit non-transient references to enclosing class instances, serializing such an inner class instance will result in serialization of its associated outer class instance as well.

Parameters:

reducer - used to reduce results of map phase

Returns:

this MapReduceTask itself

combinedWith

public MapReduceTask<KIn,VIn,KOut,VOut> combinedWith(Reducer<KOut,VOut> combiner)

Specifies Combiner to use for this MapReduceTask

Note that Reducer should not be specified as inner class. Inner classes declared in non-static contexts contain implicit non-transient references to enclosing class instances, serializing such an inner class instance will result in serialization of its associated outer class instance as well.

Parameters:

combiner - used to immediately combine results of map phase before reduce phase is invoked

Returns:

this MapReduceTask itself

Since:

5.2

timeout

public final MapReduceTask<KIn,VIn,KOut,VOut> timeout(long timeout,
                                       java.util.concurrent.TimeUnit unit)

See timeout(TimeUnit). Note: the timeout value will be converted to milliseconds and a value less or equal than zero means wait forever.

Parameters:

timeout -

unit -

Returns:

this MapReduceTask itself

timeout

public final long timeout(java.util.concurrent.TimeUnit outputTimeUnit)

Returns:

the timeout value in TimeUnit to wait for the remote map/reduce task to finish. The default value given by SyncConfiguration.replTimeout()

execute

public java.util.Map<KOut,VOut> execute()
                                 throws CacheException

Executes this task across Infinispan cluster nodes.

Returns:

a Map where each key is an output key and value is reduced value for that output key

Throws:

CacheException

distributeReducePhase

protected boolean distributeReducePhase()

useIntermediateSharedCache

protected boolean useIntermediateSharedCache()

useIntermediatePerTaskCache

protected boolean useIntermediatePerTaskCache()

executeTaskInit

protected void executeTaskInit(java.lang.String tmpCacheName)

executeMapPhase

protected java.util.Set<KOut> executeMapPhase(boolean useCompositeKeys)
                                       throws java.lang.InterruptedException,
                                              java.util.concurrent.ExecutionException

Throws:

java.lang.InterruptedException

java.util.concurrent.ExecutionException

executeMapPhaseWithLocalReduction

protected java.util.Map<KOut,VOut> executeMapPhaseWithLocalReduction()
                                                              throws java.lang.InterruptedException,
                                                                     java.util.concurrent.ExecutionException

Throws:

java.lang.InterruptedException

java.util.concurrent.ExecutionException

createTaskMapPart

protected <V> MapReduceTask.MapTaskPart<V> createTaskMapPart(org.infinispan.commands.read.MapCombineCommand<KIn,VIn,KOut,VOut> cmd,
                                                 org.infinispan.remoting.transport.Address target,
                                                 boolean distributedReduce)

executeReducePhase

protected java.util.Map<KOut,VOut> executeReducePhase(java.util.Set<KOut> allMapPhasesResponses,
                                          boolean useCompositeKeys)
                                               throws java.lang.InterruptedException,
                                                      java.util.concurrent.ExecutionException

Throws:

java.lang.InterruptedException

java.util.concurrent.ExecutionException

createReducePart

protected <V> MapReduceTask.ReduceTaskPart<V> createReducePart(org.infinispan.commands.read.ReduceCommand<KOut,VOut> cmd,
                                                   org.infinispan.remoting.transport.Address target,
                                                   java.lang.String destCacheName)

mergeResponse

private <K,V> void mergeResponse(java.util.Map<K,java.util.List<V>> result,
                       java.util.Map<K,java.util.List<V>> m)

buildMapCombineCommand

private org.infinispan.commands.read.MapCombineCommand<KIn,VIn,KOut,VOut> buildMapCombineCommand(java.lang.String taskId,
                                                                                       Mapper<KIn,VIn,KOut,VOut> m,
                                                                                       Reducer<KOut,VOut> r,
                                                                                       java.util.Collection<KIn> keys,
                                                                                       boolean reducePhaseDistributed,
                                                                                       boolean emitCompositeIntermediateKeys)

buildReduceCommand

private org.infinispan.commands.read.ReduceCommand<KOut,VOut> buildReduceCommand(java.lang.String taskId,
                                                                       java.lang.String destinationCache,
                                                                       Reducer<KOut,VOut> r,
                                                                       java.util.Collection<KOut> keys,
                                                                       boolean emitCompositeIntermediateKeys)

buildCancelCommand

private org.infinispan.commands.CancelCommand buildCancelCommand(MapReduceTask.CancellableTaskPart taskPart)

executeAsynchronously

public java.util.concurrent.Future<java.util.Map<KOut,VOut>> executeAsynchronously()

Executes this task across Infinispan cluster nodes asynchronously.

Returns:

a Future wrapping a Map where each key is an output key and value is reduced value for that output key

execute

public <R> R execute(Collator<KOut,VOut,R> collator)

Executes this task across Infinispan cluster but the final result is collated using specified Collator

Parameters:

collator - a Collator to use

Returns:

collated result

executeAsynchronously

public <R> java.util.concurrent.Future<R> executeAsynchronously(Collator<KOut,VOut,R> collator)

Executes this task asynchronously across Infinispan cluster; final result is collated using specified Collator and wrapped by Future

Parameters:

collator - a Collator to use

Returns:

collated result

aggregateReducedResult

protected void aggregateReducedResult(java.util.Map<KOut,java.util.List<VOut>> finalReduced,
                          java.util.Map<KOut,VOut> mapReceived)

mapKeysToNodes

protected <T> java.util.Map<org.infinispan.remoting.transport.Address,? extends java.util.Collection<T>> mapKeysToNodes(org.infinispan.distribution.DistributionManager dm,
                                                                                                            java.util.Collection<T> keysToMap,
                                                                                                            boolean useIntermediateCompositeKey)

mapKeysToNodes

protected <T> java.util.Map<org.infinispan.remoting.transport.Address,? extends java.util.Collection<T>> mapKeysToNodes(java.util.Collection<T> keysToMap,
                                                                                                            boolean useIntermediateCompositeKey)

mapKeysToNodes

protected <T> java.util.Map<org.infinispan.remoting.transport.Address,? extends java.util.Collection<T>> mapKeysToNodes(java.util.Collection<T> keysToMap)

clone

protected Mapper<KIn,VIn,KOut,VOut> clone(Mapper<KIn,VIn,KOut,VOut> mapper)

clone

protected Reducer<KOut,VOut> clone(Reducer<KOut,VOut> reducer)

ensureProperCacheState

private void ensureProperCacheState(AdvancedCache<KIn,VIn> cache)
                             throws java.lang.NullPointerException,
                                    java.lang.IllegalStateException

Throws:

java.lang.NullPointerException

java.lang.IllegalStateException

inputTaskKeysEmpty

protected boolean inputTaskKeysEmpty()

hashCode

public int hashCode()

Overrides:

hashCode in class java.lang.Object

equals

public boolean equals(java.lang.Object obj)

Overrides:

equals in class java.lang.Object

toString

public java.lang.String toString()

Overrides:

toString in class java.lang.Object