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 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 MapReduceTask.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

Methods

Modifier and Type	Method and Description
protected void	aggregateReducedResult(Map<KOut,List<VOut>> finalReduced, Map<KOut,VOut> mapReceived)
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<? extends Collection<T>>	mapKeysToNodes(Collection<T> keysToMap)
protected <T> Map<? extends Collection<T>>	mapKeysToNodes(Collection<T> keysToMap, boolean useIntermediateCompositeKey)
protected <T> Map<? 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 MapReduceTask.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.

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 MapReduceTask.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:

cacheConfiguration - 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

cacheConfiguration - 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 MapReduceException

Throws:

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

Throws:

MapReduceException

executeMapPhaseWithLocalReduction

protected void executeMapPhaseWithLocalReduction(Map<KOut,VOut> reducedResult)
                                          throws MapReduceException

Throws:

MapReduceException

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

Throws:

MapReduceException

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

aggregateReducedResult

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

mapKeysToNodes

protected <T> Map<? extends Collection<T>> mapKeysToNodes(DistributionManager dm,
                                              Collection<T> keysToMap,
                                              boolean useIntermediateCompositeKey)

mapKeysToNodes

protected <T> Map<? extends Collection<T>> mapKeysToNodes(Collection<T> keysToMap,
                                              boolean useIntermediateCompositeKey)

mapKeysToNodes

protected <T> Map<? 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