MapReduceTask (Infinispan JavaDoc 8.1.8.Final API)

java.lang.Object
- org.infinispan.distexec.mapreduce.MapReduceTask<KIn,VIn,KOut,VOut>

```
public class MapReduceTask<KIn,VIn,KOut,VOut>
extends 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

Field Summary

Fields
Modifier and Type	Field and Description
`protected AdvancedCache<KIn,VIn>`	`cache`
`protected List<org.infinispan.distexec.mapreduce.MapReduceTask.CancellableTaskPart>`	`cancellableTasks`
`protected CancellationService`	`cancellationService`
`protected ClusteringDependentLogic`	`clusteringDependentLogic`
`protected Reducer<KOut,VOut>`	`combiner`
`protected String`	`customIntermediateCacheName`
`static String`	`DEFAULT_TMP_CACHE_CONFIGURATION_NAME`
`protected boolean`	`distributeReducePhase`
`protected String`	`intermediateCacheConfigurationName`
`protected boolean`	`isLocalOnly`
`protected Collection<KIn>`	`keys`
`protected Mapper<KIn,VIn,KOut,VOut>`	`mapper`
`protected MapReduceManager`	`mapReduceManager`
`protected Marshaller`	`marshaller`
`protected Reducer<KOut,VOut>`	`reducer`
`protected RpcOptionsBuilder`	`rpcOptionsBuilder`
`protected 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

All Methods Instance Methods Concrete Methods
Modifier and Type	Method and Description
`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> org.infinispan.distexec.mapreduce.MapReduceTask.ReduceTaskPart<V>`	`createReducePart(ReduceCommand<KOut,VOut> cmd, Address target, String destCacheName)`
`protected <V> org.infinispan.distexec.mapreduce.MapReduceTask.MapTaskPart<V>`	`createTaskMapPart(MapCombineCommand<KIn,VIn,KOut,VOut> cmd, Address target, boolean distributedReduce)`
`protected boolean`	`distributeReducePhase()`
`boolean`	`equals(Object obj)`
`Map<KOut,VOut>`	`execute()` Executes this task across Infinispan cluster nodes.
`void`	`execute(Cache<KOut,VOut> resultsCache)` Executes this task and stores results in the provided results cache.
`<R> R`	`execute(Collator<KOut,VOut,R> collator)` Executes this task across Infinispan cluster but the final result is collated using specified `Collator`
`void`	`execute(String resultsCache)` Executes this task and stores results in the provided results cache.
`Future<Map<KOut,VOut>>`	`executeAsynchronously()` Executes this task across Infinispan cluster nodes asynchronously.
`<R> 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 Map<KOut,VOut>`	`executeHelper(String resultCache)`
`protected Set<KOut>`	`executeMapPhase()`
`protected void`	`executeMapPhaseWithLocalReduction(Map<KOut,VOut> reducedResult)`
`protected Map<KOut,VOut>`	`executeReducePhase(String resultCache, Set<KOut> allMapPhasesResponses, boolean useIntermediateSharedCache)`
`protected void`	`executeTaskInit(String tmpCacheName)`
`protected String`	`getIntermediateCacheName()`
`int`	`hashCode()`
`protected boolean`	`inputTaskKeysEmpty()`
`protected <T> Map<Address,? extends Collection<T>>`	`mapKeysToNodes(Collection<T> keysToMap)`
`protected <T> Map<Address,? extends Collection<T>>`	`mapKeysToNodes(Collection<T> keysToMap, boolean useIntermediateCompositeKey)`
`protected <T> Map<Address,? extends Collection<T>>`	`mapKeysToNodes(DistributionManager dm, Collection<T> keysToMap, boolean useIntermediateCompositeKey)`
`MapReduceTask<KIn,VIn,KOut,VOut>`	`mappedWith(Mapper<KIn,VIn,KOut,VOut> mapper)` Specifies Mapper to use for this MapReduceTask
`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, TimeUnit unit)` See `timeout(TimeUnit)`.
`long`	`timeout(TimeUnit outputTimeUnit)`
`String`	`toString()`
`protected boolean`	`useIntermediatePerTaskCache()`
`protected boolean`	`useIntermediateSharedCache()`
`MapReduceTask<KIn,VIn,KOut,VOut>`	`usingIntermediateCache(String cacheConfigurationName)` Allows this MapReduceTask to use specific intermediate custom defined cache for storage of intermediate > key/values pairs.
`MapReduceTask<KIn,VIn,KOut,VOut>`	`usingSharedIntermediateCache(String cacheName)` Allows this MapReduceTask to use a specific shared intermediate cache for storage of intermediate > key/values pairs.
`MapReduceTask<KIn,VIn,KOut,VOut>`	`usingSharedIntermediateCache(String cacheName, String cacheConfigurationName)` Allows this MapReduceTask to use a specific shared intermediate cache for storage of intermediate > key/values pairs.

Methods inherited from class java.lang.Object
clone, finalize, getClass, notify, notifyAll, wait, wait, wait

- Field Detail
  - DEFAULT_TMP_CACHE_CONFIGURATION_NAME
```
public static final 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 boolean useIntermediateSharedCache
```
  - keys
```
protected final Collection<KIn> keys
```
  - cache
```
protected final AdvancedCache<KIn,VIn> cache
```
  - marshaller
```
protected final Marshaller marshaller
```
  - mapReduceManager
```
protected final MapReduceManager mapReduceManager
```
  - cancellationService
```
protected final CancellationService cancellationService
```
  - cancellableTasks
```
protected final List<org.infinispan.distexec.mapreduce.MapReduceTask.CancellableTaskPart> cancellableTasks
```
  - taskId
```
protected final UUID taskId
```
  - clusteringDependentLogic
```
protected final ClusteringDependentLogic clusteringDependentLogic
```
  - isLocalOnly
```
protected final boolean isLocalOnly
```
  - rpcOptionsBuilder
```
protected RpcOptionsBuilder rpcOptionsBuilder
```
  - customIntermediateCacheName
```
protected String customIntermediateCacheName
```
  - intermediateCacheConfigurationName
```
protected String intermediateCacheConfigurationName
```
- 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,
                                                      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. The default timeout for this task is 0. The task will wait indefinitely for its completion.
    
    Parameters:
    
    timeout -
    
    unit -
    
    Returns:
    
    this MapReduceTask itself
  - timeout
```
public final long timeout(TimeUnit outputTimeUnit)
```
    Returns:
    
    the timeout value in TimeUnit to wait for the remote map/reduce task to finish. The default timeout is 0, the task will wait indefinitely for its completion.
  - usingIntermediateCache
```
public MapReduceTask<KIn,VIn,KOut,VOut> usingIntermediateCache(String cacheConfigurationName)
```
    Allows this MapReduceTask to use specific intermediate custom defined cache for storage of intermediate > key/values pairs. Intermediate cache is used to store output of map phase and it is not shared with other M/R tasks. Upon completion of M/R task this intermediate cache is destroyed.
    
    Parameters:
    
    cacheConfigurationName - name of the cache configuration to use for the intermediate cache
    
    Returns:
    
    this MapReduceTask iteself
    
    Since:
    
    7.0
  - usingSharedIntermediateCache
```
public MapReduceTask<KIn,VIn,KOut,VOut> usingSharedIntermediateCache(String cacheName)
```
    Allows this MapReduceTask to use a specific shared intermediate cache for storage of intermediate > key/values pairs. Intermediate shared cache is used to store output of map phase and it is shared with other M/R tasks that also specify a shared intermediate cache with the same name.
    
    Parameters:
    
    cacheName - name of the custom cache
    
    Returns:
    
    this MapReduceTask iteself
    
    Since:
    
    7.0
  - usingSharedIntermediateCache
```
public MapReduceTask<KIn,VIn,KOut,VOut> usingSharedIntermediateCache(String cacheName,
                                                                     String cacheConfigurationName)
```
    Allows this MapReduceTask to use a specific shared intermediate cache for storage of intermediate > key/values pairs. Intermediate shared cache is used to store output of map phase and it is shared with other M/R tasks that also specify a shared intermediate cache with the same name.
    Rather than using MapReduceTask default configuration for intermediate cache this method allows clients to specify custom shared cache configuration.
    
    Parameters:
    
    cacheName - name of the custom cache
    
    cacheConfigurationName - name of the cache configuration to use for the intermediate cache
    
    Returns:
    
    this MapReduceTask iteself
    
    Since:
    
    7.0
  - execute
```
public 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
  - execute
```
public void execute(Cache<KOut,VOut> resultsCache)
             throws CacheException
```
    Executes this task and stores results in the provided results cache. The results can be obtained once the execute method completes i.e., execute method is synchronous. This variant of execute method minimizes the possibility of the master JVM node exceeding its allowed maximum heap, especially if objects that are results of the reduce phase have a large memory footprint and/or multiple MapReduceTasks are executed concurrently on the master task node.
    
    Parameters:
    
    resultsCache - application provided results cache
    
    Throws:
    
    CacheException
    
    Since:
    
    7.0
  - execute
```
public void execute(String resultsCache)
             throws CacheException
```
    Executes this task and stores results in the provided results cache. The results can be obtained once the execute method completes i.e., execute method is synchronous. This variant of execute method minimizes the possibility of the master JVM node exceeding its allowed maximum heap, especially if objects that are results of the reduce phase have a large memory footprint and/or multiple MapReduceTasks are executed concurrently on the master task node.
    
    Parameters:
    
    resultsCache - application provided results cache represented by its name
    
    Throws:
    
    CacheException
    
    Since:
    
    7.0
  - executeHelper
```
protected Map<KOut,VOut> executeHelper(String resultCache)
                                throws NullPointerException,
                                       MapReduceException
```
    Throws:
    
    NullPointerException
    
    MapReduceException
  - getIntermediateCacheName
```
protected String getIntermediateCacheName()
```
  - distributeReducePhase
```
protected boolean distributeReducePhase()
```
  - useIntermediateSharedCache
```
protected boolean useIntermediateSharedCache()
```
  - useIntermediatePerTaskCache
```
protected boolean useIntermediatePerTaskCache()
```
  - executeTaskInit
```
protected void executeTaskInit(String tmpCacheName)
                        throws MapReduceException
```
    Throws:
    
    MapReduceException
  - executeMapPhase
```
protected Set<KOut> executeMapPhase()
                             throws MapReduceException,
                                    InterruptedException
```
    Throws:
    
    MapReduceException
    
    InterruptedException
  - executeMapPhaseWithLocalReduction
```
protected void executeMapPhaseWithLocalReduction(Map<KOut,VOut> reducedResult)
                                          throws MapReduceException,
                                                 InterruptedException
```
    Throws:
    
    MapReduceException
    
    InterruptedException
  - createTaskMapPart
```
protected <V> org.infinispan.distexec.mapreduce.MapReduceTask.MapTaskPart<V> createTaskMapPart(MapCombineCommand<KIn,VIn,KOut,VOut> cmd,
                                                                                               Address target,
                                                                                               boolean distributedReduce)
```
  - executeReducePhase
```
protected Map<KOut,VOut> executeReducePhase(String resultCache,
                                            Set<KOut> allMapPhasesResponses,
                                            boolean useIntermediateSharedCache)
                                     throws MapReduceException,
                                            InterruptedException
```
    Throws:
    
    MapReduceException
    
    InterruptedException
  - createReducePart
```
protected <V> org.infinispan.distexec.mapreduce.MapReduceTask.ReduceTaskPart<V> createReducePart(ReduceCommand<KOut,VOut> cmd,
                                                                                                 Address target,
                                                                                                 String destCacheName)
```
  - executeAsynchronously
```
public Future<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> 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
  - mapKeysToNodes
```
protected <T> Map<Address,? extends Collection<T>> mapKeysToNodes(DistributionManager dm,
                                                                  Collection<T> keysToMap,
                                                                  boolean useIntermediateCompositeKey)
```
  - mapKeysToNodes
```
protected <T> Map<Address,? extends Collection<T>> mapKeysToNodes(Collection<T> keysToMap,
                                                                  boolean useIntermediateCompositeKey)
```
  - mapKeysToNodes
```
protected <T> Map<Address,? extends Collection<T>> mapKeysToNodes(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)
```
  - inputTaskKeysEmpty
```
protected boolean inputTaskKeysEmpty()
```
  - hashCode
```
public int hashCode()
```
    Overrides:
    
    hashCode in class Object
  - equals
```
public boolean equals(Object obj)
```
    Overrides:
    
    equals in class Object
  - toString
```
public String toString()
```
    Overrides:
    
    toString in class Object

Class MapReduceTask<KIn,VIn,KOut,VOut>

Field Summary

Constructor Summary

Method Summary

Methods inherited from class java.lang.Object

Field Detail

DEFAULT_TMP_CACHE_CONFIGURATION_NAME

mapper

reducer

combiner

distributeReducePhase

useIntermediateSharedCache

keys

cache

marshaller

mapReduceManager

cancellationService

cancellableTasks

taskId

clusteringDependentLogic

isLocalOnly

rpcOptionsBuilder

customIntermediateCacheName

intermediateCacheConfigurationName

Constructor Detail

MapReduceTask

MapReduceTask

MapReduceTask

Method Detail

onKeys

mappedWith

reducedWith

combinedWith

timeout

timeout

usingIntermediateCache

usingSharedIntermediateCache

usingSharedIntermediateCache

execute

execute

execute

executeHelper

getIntermediateCacheName

distributeReducePhase

useIntermediateSharedCache

useIntermediatePerTaskCache

executeTaskInit

executeMapPhase

executeMapPhaseWithLocalReduction

createTaskMapPart

executeReducePhase

createReducePart

executeAsynchronously

execute

executeAsynchronously

mapKeysToNodes

mapKeysToNodes

mapKeysToNodes

clone

clone

inputTaskKeysEmpty

hashCode

equals

toString