Hi Evangelos, you might be interested in looking into a current pull request which addresses some (all?) of these issues https://github.com/infinispan/infinispan/pull/2300 Tristan On 14/02/2014 16:10, Evangelos Vazaios wrote: > Hello everyone, > > I started using the MapReduce implementation of Infinispan and I came > across some possible limitations. Thus, I want to make some suggestions > about the MapReduce (MR) implementation of Infinispan. > Depending on the algorithm, there might be some memory problems, > especially for intermediate results. > An example of such a case is group by. Suppose that we have a cluster > of 2 nodes with 2 GB available. Let a distributed cache, where simple > car objects (id,brand,colour) are stored and the total size of data is > 3.5GB. If all objects have the same colour , then all 3.5 GB would go to > only one reducer, as a result an OutOfMemoryException will be thrown. > > To overcome these limitations, I propose to add as parameter the name of > the intermediate cache to be used. This will enable the creation of a > custom configured cache that deals with the memory limitations. > > Another feature that I would like to have is to set the name of the > output cache. The reasoning behind this is similar to the one mentioned > above. > > I wait for your thoughts on these two suggestions. > > Regards, > Evangelos > _______________________________________________ > infinispan-dev mailing list > infinispan-dev(a)lists.jboss.org > https://lists.jboss.org/mailman/listinfo/infinispan-dev > > _______________________________________________ infinispan-dev mailing list infinispan-dev(a)lists.jboss.org https://lists.jboss.org/mailman/listinfo/infinispan-dev

I think that the intermediate cache is not required at all. The M/R algorithm itself can (and should!) run with memory occupied by the result of reduction. The current implementation with Map first and Reduce after that will always have these problems, using a cache for temporary caching the result is only a workaround. The only situation when temporary cache could be useful is when the result grows linearly (or close to that or even more) with the amount of reduced entries. This would be the case for groupBy producing Map<Color, List<Entry>> from all entries in cache. Then the task does not scale and should be redesigned anyway, but flushing the results into cache backed by cache store could help. Radim On 02/14/2014 04:54 PM, Vladimir Blagojevic wrote: > Tristan, > > Actually they are not addressed in this pull request but the feature > where custom output cache is used instead of results being returned is > next in the implementation pipeline. > > Evangelos, indeed, depending on a reducer function all intermediate > KOut/VOut pairs might be moved to a single node. How would custom cache > help in this case? > > Regards, > Vladimir > > > On 2/14/2014, 10:16 AM, Tristan Tarrant wrote: >> Hi Evangelos, >> >> you might be interested in looking into a current pull request which >> addresses some (all?) of these issues >> >> https://github.com/infinispan/infinispan/pull/2300 >> >> Tristan >> >> On 14/02/2014 16:10, Evangelos Vazaios wrote: >>> Hello everyone, >>> >>> I started using the MapReduce implementation of Infinispan and I came >>> across some possible limitations. Thus, I want to make some suggestions >>> about the MapReduce (MR) implementation of Infinispan. >>> Depending on the algorithm, there might be some memory problems, >>> especially for intermediate results. >>> An example of such a case is group by. Suppose that we have a cluster >>> of 2 nodes with 2 GB available. Let a distributed cache, where simple >>> car objects (id,brand,colour) are stored and the total size of data is >>> 3.5GB. If all objects have the same colour , then all 3.5 GB would go to >>> only one reducer, as a result an OutOfMemoryException will be thrown. >>> >>> To overcome these limitations, I propose to add as parameter the name of >>> the intermediate cache to be used. This will enable the creation of a >>> custom configured cache that deals with the memory limitations. >>> >>> Another feature that I would like to have is to set the name of the >>> output cache. The reasoning behind this is similar to the one mentioned >>> above. >>> >>> I wait for your thoughts on these two suggestions. >>> >>> Regards, >>> Evangelos >>> _______________________________________________ >>> infinispan-dev mailing list >>> infinispan-dev(a)lists.jboss.org >>> https://lists.jboss.org/mailman/listinfo/infinispan-dev >>> >>> >> >> >> _______________________________________________ >> infinispan-dev mailing list >> infinispan-dev(a)lists.jboss.org >> https://lists.jboss.org/mailman/listinfo/infinispan-dev > _______________________________________________ > infinispan-dev mailing list > infinispan-dev(a)lists.jboss.org > https://lists.jboss.org/mailman/listinfo/infinispan-dev -- Radim Vansa <rvansa(a)redhat.com&gt; JBoss DataGrid QA _______________________________________________ infinispan-dev mailing list infinispan-dev(a)lists.jboss.org https://lists.jboss.org/mailman/listinfo/infinispan-dev

Hi Etienne I was going to suggest using a combiner - the combiner would process the mapper results from just one node, so you should need at most double the memory on that node. I guess we could reduce the memory requirements even more if the combiner could run concurrently with the mapper... Vladimir, does it sound like a reasonable feature request?

I'm afraid in your situation using a cache store wouldn't help, as the intermediate values for the same key are stored as a list in a single entry. So if all cars are red, there would be just one intermediate key in the intermediate cache, and there would be nothing to evict to the cache store. Vladimir, do you think we could somehow "chunk" the intermediary values into multiple entries grouped by the intermediary key, to support this scenario? For reference, though, a limited version of what you're asking for is already available. You can change the configuration of the intermediary cache by defining a "__tmpMapReduce" cache in your configuration. That configuration will be used for all M/R tasks, whether they use the shared intermediate cache or they create their own.

Cheers Dan On Mon, Feb 17, 2014 at 10:18 AM, Etienne Riviere <etienne.riviere(a)unine.ch&gt; wrote: > > Hi Radim, > > I might misunderstand your suggestion but many M/R jobs actually require > to run the two phases one after the other, and henceforth to store the > intermediate results somewhere. While some may slightly reduce intermediate > memory usage by using a combiner function (e.g., the word-count example), I > don’t see how we can avoid intermediate storage altogether. > > Thanks, > Etienne (leads project — as Evangelos who initiated the thread) > > On 17 Feb 2014, at 08:48, Radim Vansa <rvansa(a)redhat.com&gt; wrote: > > > I think that the intermediate cache is not required at all. The M/R > > algorithm itself can (and should!) run with memory occupied by the > > result of reduction. The current implementation with Map first and > > Reduce after that will always have these problems, using a cache for > > temporary caching the result is only a workaround. > > > > The only situation when temporary cache could be useful is when the > > result grows linearly (or close to that or even more) with the amount of > > reduced entries. This would be the case for groupBy producing Map<Color, > > List<Entry>> from all entries in cache. Then the task does not scale and > > should be redesigned anyway, but flushing the results into cache backed > > by cache store could help. > > > > Radim > > > > On 02/14/2014 04:54 PM, Vladimir Blagojevic wrote: > >> Tristan, > >> > >> Actually they are not addressed in this pull request but the feature > >> where custom output cache is used instead of results being returned is > >> next in the implementation pipeline. > >> > >> Evangelos, indeed, depending on a reducer function all intermediate > >> KOut/VOut pairs might be moved to a single node. How would custom cache > >> help in this case? > >> > >> Regards, > >> Vladimir > >> > >> > >> On 2/14/2014, 10:16 AM, Tristan Tarrant wrote: > >>> Hi Evangelos, > >>> > >>> you might be interested in looking into a current pull request which > >>> addresses some (all?) of these issues > >>> > >>> https://github.com/infinispan/infinispan/pull/2300 > >>> > >>> Tristan > >>> > >>> On 14/02/2014 16:10, Evangelos Vazaios wrote: > >>>> Hello everyone, > >>>> > >>>> I started using the MapReduce implementation of Infinispan and I came > >>>> across some possible limitations. Thus, I want to make some > >>>> suggestions > >>>> about the MapReduce (MR) implementation of Infinispan. > >>>> Depending on the algorithm, there might be some memory problems, > >>>> especially for intermediate results. > >>>> An example of such a case is group by. Suppose that we have a > >>>> cluster > >>>> of 2 nodes with 2 GB available. Let a distributed cache, where > >>>> simple > >>>> car objects (id,brand,colour) are stored and the total size of data > >>>> is > >>>> 3.5GB. If all objects have the same colour , then all 3.5 GB would go > >>>> to > >>>> only one reducer, as a result an OutOfMemoryException will be thrown. > >>>> > >>>> To overcome these limitations, I propose to add as parameter the name > >>>> of > >>>> the intermediate cache to be used. This will enable the creation of a > >>>> custom configured cache that deals with the memory limitations. > >>>> > >>>> Another feature that I would like to have is to set the name of the > >>>> output cache. The reasoning behind this is similar to the one > >>>> mentioned > >>>> above. > >>>> > >>>> I wait for your thoughts on these two suggestions. > >>>> > >>>> Regards, > >>>> Evangelos > >>>> _______________________________________________ > >>>> infinispan-dev mailing list > >>>> infinispan-dev(a)lists.jboss.org > >>>> https://lists.jboss.org/mailman/listinfo/infinispan-dev > >>>> > >>>> > >>> > >>> > >>> _______________________________________________ > >>> infinispan-dev mailing list > >>> infinispan-dev(a)lists.jboss.org > >>> https://lists.jboss.org/mailman/listinfo/infinispan-dev > >> _______________________________________________ > >> infinispan-dev mailing list > >> infinispan-dev(a)lists.jboss.org > >> https://lists.jboss.org/mailman/listinfo/infinispan-dev > > > > > > -- > > Radim Vansa <rvansa(a)redhat.com&gt; > > JBoss DataGrid QA > > > > _______________________________________________ > > infinispan-dev mailing list > > infinispan-dev(a)lists.jboss.org > > https://lists.jboss.org/mailman/listinfo/infinispan-dev > > > _______________________________________________ > infinispan-dev mailing list > infinispan-dev(a)lists.jboss.org > https://lists.jboss.org/mailman/listinfo/infinispan-dev _______________________________________________ infinispan-dev mailing list infinispan-dev(a)lists.jboss.org https://lists.jboss.org/mailman/listinfo/infinispan-dev

Hi Radim, I might misunderstand your suggestion but many M/R jobs actually require to run the two phases one after the other, and henceforth to store the intermediate results somewhere. While some may slightly reduce intermediate memory usage by using a combiner function (e.g., the word-count example), I don’t see how we can avoid intermediate storage altogether. Thanks, Etienne (leads project — as Evangelos who initiated the thread) On 17 Feb 2014, at 08:48, Radim Vansa <rvansa(a)redhat.com&gt; wrote: > I think that the intermediate cache is not required at all. The M/R > algorithm itself can (and should!) run with memory occupied by the > result of reduction. The current implementation with Map first and > Reduce after that will always have these problems, using a cache for > temporary caching the result is only a workaround. > > The only situation when temporary cache could be useful is when the > result grows linearly (or close to that or even more) with the amount of > reduced entries. This would be the case for groupBy producing Map<Color, > List<Entry>> from all entries in cache. Then the task does not scale and > should be redesigned anyway, but flushing the results into cache backed > by cache store could help. > > Radim > > On 02/14/2014 04:54 PM, Vladimir Blagojevic wrote: >> Tristan, >> >> Actually they are not addressed in this pull request but the feature >> where custom output cache is used instead of results being returned is >> next in the implementation pipeline. >> >> Evangelos, indeed, depending on a reducer function all intermediate >> KOut/VOut pairs might be moved to a single node. How would custom cache >> help in this case? >> >> Regards, >> Vladimir >> >> >> On 2/14/2014, 10:16 AM, Tristan Tarrant wrote: >>> Hi Evangelos, >>> >>> you might be interested in looking into a current pull request which >>> addresses some (all?) of these issues >>> >>> https://github.com/infinispan/infinispan/pull/2300 >>> >>> Tristan >>> >>> On 14/02/2014 16:10, Evangelos Vazaios wrote: >>>> Hello everyone, >>>> >>>> I started using the MapReduce implementation of Infinispan and I came >>>> across some possible limitations. Thus, I want to make some suggestions >>>> about the MapReduce (MR) implementation of Infinispan. >>>> Depending on the algorithm, there might be some memory problems, >>>> especially for intermediate results. >>>> An example of such a case is group by. Suppose that we have a cluster >>>> of 2 nodes with 2 GB available. Let a distributed cache, where simple >>>> car objects (id,brand,colour) are stored and the total size of data is >>>> 3.5GB. If all objects have the same colour , then all 3.5 GB would go to >>>> only one reducer, as a result an OutOfMemoryException will be thrown. >>>> >>>> To overcome these limitations, I propose to add as parameter the name of >>>> the intermediate cache to be used. This will enable the creation of a >>>> custom configured cache that deals with the memory limitations. >>>> >>>> Another feature that I would like to have is to set the name of the >>>> output cache. The reasoning behind this is similar to the one mentioned >>>> above. >>>> >>>> I wait for your thoughts on these two suggestions. >>>> >>>> Regards, >>>> Evangelos >>>> _______________________________________________ >>>> infinispan-dev mailing list >>>> infinispan-dev(a)lists.jboss.org >>>> https://lists.jboss.org/mailman/listinfo/infinispan-dev >>>> >>>> >>> >>> _______________________________________________ >>> infinispan-dev mailing list >>> infinispan-dev(a)lists.jboss.org >>> https://lists.jboss.org/mailman/listinfo/infinispan-dev >> _______________________________________________ >> infinispan-dev mailing list >> infinispan-dev(a)lists.jboss.org >> https://lists.jboss.org/mailman/listinfo/infinispan-dev > > -- > Radim Vansa <rvansa(a)redhat.com&gt; > JBoss DataGrid QA > > _______________________________________________ > infinispan-dev mailing list > infinispan-dev(a)lists.jboss.org > https://lists.jboss.org/mailman/listinfo/infinispan-dev _______________________________________________ infinispan-dev mailing list infinispan-dev(a)lists.jboss.org https://lists.jboss.org/mailman/listinfo/infinispan-dev

I think Hadoop only loads a block of intermediate values in memory at once, and can even sort the intermediate values (with a user-supplied comparison function) so that the reduce function can work on a sorted list without loading the values in memory itself.

Hi Radim, Since Hadoop is the most popular implementation of MapReduce I will give a brief overview of how it works and then I'll provide with an example where the reducers must run over the whole list of values with the same key. Hadoop MR overview. MAP 1) Input file(s) are split into pieces of 64MB 2) For each split hadoop creates one map task and then assign the task to a cluster node 3) The splits are read as key,value pairs and the map function of Mapper is called. The mapper can output arbitrary number of intermediate key,value pairs 4) the output from the mapper is stored in a buffer in memory. After a certain threshold is reached the pairs are sorted by key and if there is a combiner it is run on the pairs that have the same key. Then, the output is flushed on the HDFS.

SHUFFLE hadoop decides the Reducer that should process each key by running a partitioner. The default partitioner decides with the following way: reducer = intermidKey.hashCode() % numberOfReducer Finally, the intermediate key,value pairs are sent to the reducers

REDUCE 1) Reducer sorts all key,value pairs by key and then groups the values with the same key. As a result reducers receive their keys sorted.

2) for each Key,List<Value> the reduce function of the reducer is called. Reducer can also emit arbitrary number of key,value pairs

Additionally, hadoop lets you customize almost every aspect of the code run from how the input is split and read as key value pairs to how it is partitioned and sorted.

A simple example is group by and computing an average over the grouped values. Let the dataset be webpages (url,domain,sentiment) and we want to compute the average sentiment for each domain in the dataset then the mapper for each webpages wp. will run map(wp.url,wp): emit(wp.domain,wp.sentiment) and in reducer: reduce(domain,Iterable<Double> values): counter = 0 sum = 0 while(values.hasNext()) counter++; sum += values.next() emit(domain,sum/counter) I know that this approach is not optimized. But, I wanted give a simple example.

Dan, only the the values for one intermediate key must be in memory? or all the intermediate key,value pairs that are assigned to one reducer must be in memory?

Cheers, Evangelos On 02/18/2014 11:59 AM, Dan Berindei wrote: > Radim, this is how our M/R algorithm works (Hadoop may do it differently): > > * The mapping phase generates a Map<IntKey, Collection<IntValue>> on each > node (Int meaning intermediate). > * In the combine (local reduce) phase, a combine operation takes as input > an IntKey and a Collection<IntValue> with only the values that were > produced on that node. > * In the (global) reduce phase, all the intermediate values for each key > are merged, and a reduce operation takes an intermediate key and a sequence > of *all* the intermediate values generated for that key. These reduce > operations are completely independent, so each intermediate key can be > mapped to a different node (distributed reduce), while still having access > to all the intermediate values at once. > * In the end, the collator takes the Map<IntKey, IntValue> from the reduce > phase and produces a single value. > > If a combiner can be used, then I believe it can also be run in parallel > with a LinkedBlockingQueue between the mapper and the combiner. But > sometimes the reduce algorithm can only be run on the entire collection of > values (e.g if you want to find the median, or a percentile). > > The limitation we have now is that in the reduce phase, the entire list of > values for one intermediate key must be in memory at once. I think Hadoop > only loads a block of intermediate values in memory at once, and can even > sort the intermediate values (with a user-supplied comparison function) so > that the reduce function can work on a sorted list without loading the > values in memory itself. > > Cheers > Dan

On 02/18/2014 01:40 PM, Dan Berindei wrote: > On Tue, Feb 18, 2014 at 12:21 PM, Evangelos Vazaios <vagvaz(a)gmail.com&gt;wrote: > >> Hi Radim, >> >> Since Hadoop is the most popular implementation of MapReduce I will give >> a brief overview of how it works and then I'll provide with an example >> where the reducers must run over the whole list of values with the same >> key. >> >> Hadoop MR overview. >> >> MAP >> >> 1) Input file(s) are split into pieces of 64MB >> 2) For each split hadoop creates one map task and then assign the task >> to a cluster node >> 3) The splits are read as key,value pairs and the map function of Mapper >> is called. The mapper can output arbitrary number of intermediate >> key,value pairs >> 4) the output from the mapper is stored in a buffer in memory. After a >> certain threshold is reached the pairs are sorted by key and if there is >> a combiner it is run on the pairs that have the same key. Then, the >> output is flushed on the HDFS. >> > Ok, so Hadoop runs the combiner more or less concurrently with the mappers. > > I'm curious if there are any M/R tasks that benefit from the sorting the > keys here, we just put the intermediate values in a Map<IK, Coll<IV>>. We > could do about the same by passing this map (or rather each entry in the > map) to the combiner when it reaches a certain threshold, but I'm not > convinced about the need to sort it. > Well there are algorithms that make use of it. Implementing a graph algorithm can take use of it.Where the graph is split into k partitions and each partition is assigned to one Mapper and Reducer. Mappers compute the outgoing messages and output them to reducers. Then, reducers can read the partition file sequentially to update the vertices. This is just one use case that came to my mind. >> SHUFFLE >> >> hadoop decides the Reducer that should process each key by running a >> partitioner. The default partitioner decides with the following way: >> reducer = intermidKey.hashCode() % numberOfReducer >> Finally, the intermediate key,value pairs are sent to the reducers >> > Is this algorithm set in stone, in that some M/R tasks rely on it? In our > impl, the user could use grouping to direct a set of intermediate keys to > the same node for reducing, but otherwise the reducing node is more or less > random. > The default partitioner does exactly that check the actual code for hadoop 1.2.1 here http://goo.gl/he9yHO >> REDUCE >> >> 1) Reducer sorts all key,value pairs by key and then groups the values >> with the same key. As a result reducers receive their keys sorted. >> > I guess this sorting is only relevant if the reduce phase happens on a > single thread, on a single node? If the reduce happens in parallel, the > ordering is going to be lost anyway. Each reduce task is run on a single thread, but you can run more than one reduce tasks on a given node. The key ordering will not be lost. The values are not ordered in any way. Moreover, the call to the reducer is reduce(Key key, Iterable<Value> values) I cannot think of a way that the order is lost. > >> 2) for each Key,List<Value> the reduce function of the reducer is >> called. Reducer can also emit arbitrary number of key,value pairs >> > We limit the reducer (and the combiner) to emit a single value, which is > paired with the input key. We may need to lift this restriction, if only to > make porting/adapting tasks easier. > > >> Additionally, hadoop lets you customize almost every aspect of the code >> run from how the input is split and read as key value pairs to how it is >> partitioned and sorted. >> > Does that mean you can sort the values as well? I was thinking of each > reduce() call as independent, and then only the order of values for one > intermediate key would be relevant. I guess some tasks may require keeping > state across all the reduce() calls and then the order of key matters, but > then the reduce phase can't be parallelized, either across the cluster or > on a single node. I was not very clear here. You can set the partitioner for a specific job. You may also set the key comparator, as a result change the way that intermediate keys are sorted. Additionally, one can change how keys are grouped into one reduce call by setting the GroupComparator class. A simple example would be to have sales(date,amount) and you want to create totals for each month of the year. so for the key: (year,month) and value: amount. by overriding the keyClass hashCode function you can send all the intermediate pairs with the same year to the same reducer and then you can set the groupComparator to group together all the values with the same year. Cheers, Evangelos >> A simple example is group by and computing an average over the grouped >> values. Let the dataset be webpages (url,domain,sentiment) and we want >> to compute the average sentiment for each domain in the dataset then the >> mapper for each webpages wp. will run >> map(wp.url,wp): >> emit(wp.domain,wp.sentiment) >> >> and in reducer: >> reduce(domain,Iterable<Double> values): >> counter = 0 >> sum = 0 >> while(values.hasNext()) >> counter++; >> sum += values.next() >> emit(domain,sum/counter) >> >> I know that this approach is not optimized. But, I wanted give a simple >> example. >> > I think it can also be optimized to use a combiner, if we emit a (domain, > counter, sum) tuple :) > > >> Dan, only the the values for one intermediate key must be in memory? or >> all the intermediate key,value pairs that are assigned to one reducer >> must be in memory? >> > With the default configuration, all the key/value pairs assigned to one > reducer must be in memory. But one can define the __tmpMapReduce cache in > the configuration and configure eviction with a cache store (note that > because of how our eviction works, the actual container size is at least > concurrencyLevel rounded up to the next power of 2). The problem is that > there is only one configuration for all the M/R tasks [1]. > > Note that because we only run the combiner after the mapping phase is > complete, we do need to keep in memory all the results of the mapping phase > from that node (those are not stored in a cache). I've created an issue in > JIRA for this [2]. > > Cheers > Dan > > [1] https://issues.jboss.org/browse/ISPN-4021 > [2] https://issues.jboss.org/browse/ISPN-4022 > > > >> Cheers, >> Evangelos >> >> On 02/18/2014 11:59 AM, Dan Berindei wrote: >>> Radim, this is how our M/R algorithm works (Hadoop may do it >> differently): >>> * The mapping phase generates a Map<IntKey, Collection<IntValue>> on each >>> node (Int meaning intermediate). >>> * In the combine (local reduce) phase, a combine operation takes as input >>> an IntKey and a Collection<IntValue> with only the values that were >>> produced on that node. >>> * In the (global) reduce phase, all the intermediate values for each key >>> are merged, and a reduce operation takes an intermediate key and a >> sequence >>> of *all* the intermediate values generated for that key. These reduce >>> operations are completely independent, so each intermediate key can be >>> mapped to a different node (distributed reduce), while still having >> access >>> to all the intermediate values at once. >>> * In the end, the collator takes the Map<IntKey, IntValue> from the >> reduce >>> phase and produces a single value. >>> >>> If a combiner can be used, then I believe it can also be run in parallel >>> with a LinkedBlockingQueue between the mapper and the combiner. But >>> sometimes the reduce algorithm can only be run on the entire collection >> of >>> values (e.g if you want to find the median, or a percentile). >>> >>> The limitation we have now is that in the reduce phase, the entire list >> of >>> values for one intermediate key must be in memory at once. I think Hadoop >>> only loads a block of intermediate values in memory at once, and can even >>> sort the intermediate values (with a user-supplied comparison function) >> so >>> that the reduce function can work on a sorted list without loading the >>> values in memory itself. >>> >>> Cheers >>> Dan > > > _______________________________________________ > infinispan-dev mailing list > infinispan-dev(a)lists.jboss.org > https://lists.jboss.org/mailman/listinfo/infinispan-dev > _______________________________________________ infinispan-dev mailing list infinispan-dev(a)lists.jboss.org https://lists.jboss.org/mailman/listinfo/infinispan-dev

On Tue, Feb 18, 2014 at 2:17 PM, Evangelos Vazaios <vagvaz(a)gmail.com&gt; wrote: > On 02/18/2014 01:40 PM, Dan Berindei wrote: >> On Tue, Feb 18, 2014 at 12:21 PM, Evangelos Vazaios <vagvaz(a)gmail.com >> wrote: >> >>> Hi Radim, >>> >>> Since Hadoop is the most popular implementation of MapReduce I will give >>> a brief overview of how it works and then I'll provide with an example >>> where the reducers must run over the whole list of values with the same >>> key. >>> >>> Hadoop MR overview. >>> >>> MAP >>> >>> 1) Input file(s) are split into pieces of 64MB >>> 2) For each split hadoop creates one map task and then assign the task >>> to a cluster node >>> 3) The splits are read as key,value pairs and the map function of Mapper >>> is called. The mapper can output arbitrary number of intermediate >>> key,value pairs >>> 4) the output from the mapper is stored in a buffer in memory. After a >>> certain threshold is reached the pairs are sorted by key and if there is >>> a combiner it is run on the pairs that have the same key. Then, the >>> output is flushed on the HDFS. >>> >> >> Ok, so Hadoop runs the combiner more or less concurrently with the > mappers. >> >> I'm curious if there are any M/R tasks that benefit from the sorting the >> keys here, we just put the intermediate values in a Map<IK, Coll<IV>>. We >> could do about the same by passing this map (or rather each entry in the >> map) to the combiner when it reaches a certain threshold, but I'm not >> convinced about the need to sort it. >> > Well there are algorithms that make use of it. Implementing a graph > algorithm can take use of it.Where the graph is split into k partitions > and each partition is assigned to one Mapper and Reducer. Mappers > compute the outgoing messages and output them to reducers. Then, > reducers can read the partition file sequentially to update the > vertices. This is just one use case that came to my mind. > I thought the partitioning only happens during the shuffle phase, and mappers/combiners don't know about partitions at all? I understand that reducers may need the intermediary keys to be sorted, I'm asking about the combiners, since even if the keys from one block are sorted, the complete list of keys they receive is not sorted (unless a new combiner is created for each input block).

> >> >>> SHUFFLE >>> >>> hadoop decides the Reducer that should process each key by running a >>> partitioner. The default partitioner decides with the following way: >>> reducer = intermidKey.hashCode() % numberOfReducer >>> Finally, the intermediate key,value pairs are sent to the reducers >>> >> >> Is this algorithm set in stone, in that some M/R tasks rely on it? In our >> impl, the user could use grouping to direct a set of intermediate keys to >> the same node for reducing, but otherwise the reducing node is more or > less >> random. >> > The default partitioner does exactly that check the actual code for > hadoop 1.2.1 here > http://goo.gl/he9yHO > So API documentation doesn't specify it, but users still rely on this particular behaviour? BTW, is there always one reducer one each node, or can there be multiple reducers on each node? If it's the latter, it should be relatively easy to model this in Infinispan using grouping. If it's the former, I'm not so sure...

>> >>> REDUCE >>> >>> 1) Reducer sorts all key,value pairs by key and then groups the values >>> with the same key. As a result reducers receive their keys sorted. >>> >> >> I guess this sorting is only relevant if the reduce phase happens on a >> single thread, on a single node? If the reduce happens in parallel, the >> ordering is going to be lost anyway. > Each reduce task is run on a single thread, but you can run more than > one reduce tasks on a given node. The key ordering will not be lost. The > values are not ordered in any way. Moreover, the call to the reducer is > reduce(Key key, Iterable<Value> values) I cannot think of a way that the > order is lost. >> > Right, the call to the reducer is with a single key, but I'm assuming the order of the calls matters (e.g. because the reduces keeps some internal state across reduce() calls), otherwise there's no point in sorting the keys. Calling the same reducer from multiple threads (like we do) would definitely mess up the order of the calls. ATM we only have one reducer per node, which can be called from multiple threads, but it shouldn't be too hard to allow multiple reducers per node and to run each of them in a single thread.

>> >>> 2) for each Key,List<Value> the reduce function of the reducer is >>> called. Reducer can also emit arbitrary number of key,value pairs >>> >> >> We limit the reducer (and the combiner) to emit a single value, which is >> paired with the input key. We may need to lift this restriction, if only > to >> make porting/adapting tasks easier. >> >> >>> >>> Additionally, hadoop lets you customize almost every aspect of the code >>> run from how the input is split and read as key value pairs to how it is >>> partitioned and sorted. >>> >> >> Does that mean you can sort the values as well? I was thinking of each >> reduce() call as independent, and then only the order of values for one >> intermediate key would be relevant. I guess some tasks may require > keeping >> state across all the reduce() calls and then the order of key matters, > but >> then the reduce phase can't be parallelized, either across the cluster or >> on a single node. > > I was not very clear here. You can set the partitioner for a specific > job. You may also set the key comparator, as a result change the way > that intermediate keys are sorted. Additionally, one can change how keys > are grouped into one reduce call by setting the GroupComparator class. A > simple example would be to have sales(date,amount) and you want to > create totals for each month of the year. > so for the key: (year,month) and value: amount. > by overriding the keyClass hashCode function you can send all the > intermediate pairs with the same year to the same reducer > > and then you can set the groupComparator to group together all the > values with the same year. > You mean set the groupComparator to group together all the values with the same month? I don't think so, because the key is already (year, month). But if you wanted to collect the totals for each year you could just use the year as the intermediary key. So I don't quite understand how your example is supposed to work.

Besides, each reduce() call receives just one key, if you have keys (2013, 1) and (2013, 2) and the groupComparator decides they should map to the same group, which key does the reducer see? I think a regular equals() should be good enough for us here, since we already need equals() in order to put the intermediary keys in the intermediary cache.

Cheers Dan

> > Cheers, > Evangelos > > >> >>> A simple example is group by and computing an average over the grouped >>> values. Let the dataset be webpages (url,domain,sentiment) and we want >>> to compute the average sentiment for each domain in the dataset then the >>> mapper for each webpages wp. will run >>> map(wp.url,wp): >>> emit(wp.domain,wp.sentiment) >>> >>> and in reducer: >>> reduce(domain,Iterable<Double> values): >>> counter = 0 >>> sum = 0 >>> while(values.hasNext()) >>> counter++; >>> sum += values.next() >>> emit(domain,sum/counter) >>> >>> I know that this approach is not optimized. But, I wanted give a simple >>> example. >>> >> >> I think it can also be optimized to use a combiner, if we emit a (domain, >> counter, sum) tuple :) > >> >> >> >>> Dan, only the the values for one intermediate key must be in memory? or >>> all the intermediate key,value pairs that are assigned to one reducer >>> must be in memory? >>> >> >> With the default configuration, all the key/value pairs assigned to one >> reducer must be in memory. But one can define the __tmpMapReduce cache in >> the configuration and configure eviction with a cache store (note that >> because of how our eviction works, the actual container size is at least >> concurrencyLevel rounded up to the next power of 2). The problem is that >> there is only one configuration for all the M/R tasks [1]. >> >> Note that because we only run the combiner after the mapping phase is >> complete, we do need to keep in memory all the results of the mapping > phase >> from that node (those are not stored in a cache). I've created an issue > in >> JIRA for this [2]. >> >> Cheers >> Dan >> >> [1] https://issues.jboss.org/browse/ISPN-4021 >> [2] https://issues.jboss.org/browse/ISPN-4022 >> >> >> >>> Cheers, >>> Evangelos >>> >>> On 02/18/2014 11:59 AM, Dan Berindei wrote: >>>> Radim, this is how our M/R algorithm works (Hadoop may do it >>> differently): >>>> >>>> * The mapping phase generates a Map<IntKey, Collection<IntValue>> on > each >>>> node (Int meaning intermediate). >>>> * In the combine (local reduce) phase, a combine operation takes as > input >>>> an IntKey and a Collection<IntValue> with only the values that were >>>> produced on that node. >>>> * In the (global) reduce phase, all the intermediate values for each > key >>>> are merged, and a reduce operation takes an intermediate key and a >>> sequence >>>> of *all* the intermediate values generated for that key. These reduce >>>> operations are completely independent, so each intermediate key can be >>>> mapped to a different node (distributed reduce), while still having >>> access >>>> to all the intermediate values at once. >>>> * In the end, the collator takes the Map<IntKey, IntValue> from the >>> reduce >>>> phase and produces a single value. >>>> >>>> If a combiner can be used, then I believe it can also be run in > parallel >>>> with a LinkedBlockingQueue between the mapper and the combiner. But >>>> sometimes the reduce algorithm can only be run on the entire collection >>> of >>>> values (e.g if you want to find the median, or a percentile). >>>> >>>> The limitation we have now is that in the reduce phase, the entire list >>> of >>>> values for one intermediate key must be in memory at once. I think > Hadoop >>>> only loads a block of intermediate values in memory at once, and can > even >>>> sort the intermediate values (with a user-supplied comparison function) >>> so >>>> that the reduce function can work on a sorted list without loading the >>>> values in memory itself. >>>> >>>> Cheers >>>> Dan >>> >> >> >> >> _______________________________________________ >> infinispan-dev mailing list >> infinispan-dev(a)lists.jboss.org >> https://lists.jboss.org/mailman/listinfo/infinispan-dev >> > > _______________________________________________ > infinispan-dev mailing list > infinispan-dev(a)lists.jboss.org > https://lists.jboss.org/mailman/listinfo/infinispan-dev > _______________________________________________ infinispan-dev mailing list infinispan-dev(a)lists.jboss.org https://lists.jboss.org/mailman/listinfo/infinispan-dev

Sorry, I didn't get too much from that example either, I gave up after the second "registering is fun" popup :) One last question: with Hadoop I imagine it's quite easy to leave the results of the M/R job on the distributed FS and start a new job to M/R from that. Do you think it would be important to offer something similar in Infinispan (i.e. put the result of the reducers in a cache instead of returning it to the user)?

This is on our todo list https://issues.jboss.org/browse/ISPN-4002 Cool, I thought I saw it somewhere but I didn't get to actually search in JIRA for it :) Vladimir, what do you think about the partitioning/sorting/grouping stuff? I'm not sure if it should be a priority for us: there are certainly Hadoop jobs that use those and would be pretty tricky to translate to our API, but on the other hand I'm sure most jobs are ok with an unordered Map<K, V> as the output. Cheers Dan

The limitation we have now is that in the reduce phase, the entire list of values for one intermediate key must be in memory at once. I think Hadoop only loads a block of intermediate values in memory at once, and can even sort the intermediate values (with a user-supplied comparison function) so that the reduce function can work on a sorted list without loading the values in memory itself.

On 02/18/2014 05:36 PM, Vladimir Blagojevic wrote: > On 2/18/2014, 4:59 AM, Dan Berindei wrote: >> >> The limitation we have now is that in the reduce phase, the entire >> list of values for one intermediate key must be in memory at once. I >> think Hadoop only loads a block of intermediate values in memory at >> once, and can even sort the intermediate values (with a user-supplied >> comparison function) so that the reduce function can work on a sorted >> list without loading the values in memory itself. >> >> > Dan and others, > > This is where Sanne's idea comes into play. Why collect entire list of > intermediate values for each intermediate key and then invoke reduce on > those values when we can invoke reduce each time new intermediate value > gets inserted? > Because you cant. What you are saying is more like combining than reducing. If there is a combiner in the MapReduceTask you can execute the combiner on a subset (in your case 2) values with the same key and output one. But, this is not possible always.

> https://issues.jboss.org/browse/ISPN-3999 > > Cheers, > Vladimir > _______________________________________________ > infinispan-dev mailing list > infinispan-dev(a)lists.jboss.org > https://lists.jboss.org/mailman/listinfo/infinispan-dev > _______________________________________________ infinispan-dev mailing list infinispan-dev(a)lists.jboss.org https://lists.jboss.org/mailman/listinfo/infinispan-dev