4:19 p.m.
Hey guys,
Before moving forward with next iteration of map reduce I wanted to hear
your thoughts about the following proposal. After we agree on the
general direction I will transcribe the agreed design on a wiki page and
start implementation.
Shortcoming of current map reduce implementation
While our current map reduce implementation is more than a proof of a
concept there are several drawbacks preventing it from being an
industrial grade map reduce solution. The main drawback is the inability
of the current solution to deal with a large data (in GB/TB) map reduce
problems. This shortcoming is mainly around our reduce phase execution.
Reduce phase, as you might know, is currently done on a single
Infinispan master task node; reduce phase of map reduce problems we can
support (data size wise) is therefore shrunk to a working memory of a
single node.
Proposed solution
The proposed solution involves distributing execution of reduce phase
tasks across the cluster thus effectively achieving higher reduce task
parallelization and at the same time removing the above mentioned reduce
phase restriction. Effectively leveraging our consistent hashing
solution even further we can parallelize reduce phase and elevate our
map reduce solution to an industrial level. Here is how we can achieve that.
Map phase
MapReduceTask, as it currently does, will hash task input keys and group
them by execution node N they are hashed to. For each node N and its
grouped input KIn keys MapReduceTask creates a MapCombineCommand which
is migrated to an execution target node N. MapCombineCommand is similar
to current MapReduceCommand. MapCombineCommand takes an instance of a
Mapper and an instance of a Reducer, which is a combiner [1].
Once loaded into target execution node MapCombineCommand takes each
local KIn key and executes Mapper method void map(KIn key, VIn value,
Collector<KOut, VOut> collector). Results are collected to a common
Collector<KOut, VOut> collector and combine phase is initiated. A
Combiner, if specified, takes KOut keys and imediatelly invokes reduce
phase on keys. The result of mapping phase executed on each node is
<KOut, VOut> map M. There will be one resulting M map per execution node N.
At the end of combine phase instead of returning map M to the master
task node (as we currently do), we now hash each KOut in map M and group
KOut keys by the execution node N they are hashed to. Each group of KOut
keys and its VOut values, hashed to the same node, is wrapped with a new
command Migrate. Command Migrate, which is very similar to
PutKeyValueCommand,executed on Infinispan target node N esentially
maintains KOut K -> List<VOut> mapping, i.e all KOut/VOut pairs from all
executed MapCombineCommands will be collocated on a node N where KOut is
hashed to and value for KOut will be a list of all VOut values. We
essentially collect all VOut values under each KOut for all executed
MapCombineCommands.
At this point MapCombineCommand has finished its execution; list of KOut
keys is returned to a master node and its MapReduceTask. We do not
return VOut values as we do not need them at master task node.
MapReduceTask is ready to start with reduce phase.
Reduce phase
MapReduceTask initializes ReduceCommand with a user specified Reducer.
For each key KOut collected from a map phase we group them by execution
node N they are hashed to. For each node N and its grouped input KOut
keys MapReduceTask creates a ReduceCommand and sends it to a node N
where KOut keys are hashed. Once loaded on target execution node,
ReduceCommand for each KOut key grabs list of values VOut and invokes:
VOut reduce(KOut reducedKey, Iterator<VOut> iter).
A result of ReduceCommand is a map M where each key is KOut and value is
VOut. Each Infinispan execution node N returns one map M where each key
KOut is hashed to N and each VOut is KOut's reduced value.
When all ReduceCommands return to a calling node, MapReduceTask simply
combines all these M maps and returns final Map<KOut, VOut> as a result
of MapReduceTask. All intermediate KOut->List<VOut> maps left on
Infinispan cluster are then cleaned up.
[1] See section 4.3 of http://research.google.com/archive/mapreduce.html
7:43 p.m.
On 14 Feb 2012, at 15:19, Vladimir Blagojevic wrote:
Hey guys,
Before moving forward with next iteration of map reduce I wanted to hear
your thoughts about the following proposal. After we agree on the
general direction I will transcribe the agreed design on a wiki page and
start implementation.
Shortcoming of current map reduce implementation
While our current map reduce implementation is more than a proof of a
concept there are several drawbacks preventing it from being an
industrial grade map reduce solution. The main drawback is the inability
of the current solution to deal with a large data (in GB/TB) map reduce
problems. This shortcoming is mainly around our reduce phase execution.
Reduce phase, as you might know, is currently done on a single
Infinispan master task node; reduce phase of map reduce problems we can
support (data size wise) is therefore shrunk to a working memory of a
single node.
I was under the impression reduce is distributed too? Don't we do the mapping on each
node, then a first-pass reduce on each node too, before streaming results back to the
caller node?
Proposed solution
The proposed solution involves distributing execution of reduce phase
tasks across the cluster thus effectively achieving higher reduce task
parallelization and at the same time removing the above mentioned reduce
phase restriction. Effectively leveraging our consistent hashing
solution even further we can parallelize reduce phase and elevate our
map reduce solution to an industrial level. Here is how we can achieve that.
Map phase
MapReduceTask, as it currently does, will hash task input keys and group
them by execution node N they are hashed to. For each node N and its
grouped input KIn keys MapReduceTask creates a MapCombineCommand which
is migrated to an execution target node N. MapCombineCommand is similar
to current MapReduceCommand. MapCombineCommand takes an instance of a
Mapper and an instance of a Reducer, which is a combiner [1].
Once loaded into target execution node MapCombineCommand takes each
local KIn key and executes Mapper method void map(KIn key, VIn value,
Collector<KOut, VOut> collector). Results are collected to a common
Collector<KOut, VOut> collector and combine phase is initiated. A
Combiner, if specified, takes KOut keys and imediatelly invokes reduce
phase on keys. The result of mapping phase executed on each node is
<KOut, VOut> map M. There will be one resulting M map per execution node N.
At the end of combine phase instead of returning map M to the master
task node (as we currently do), we now hash each KOut in map M and group
KOut keys by the execution node N they are hashed to. Each group of KOut
keys and its VOut values, hashed to the same node, is wrapped with a new
command Migrate. Command Migrate, which is very similar to
PutKeyValueCommand,executed on Infinispan target node N esentially
maintains KOut K -> List<VOut> mapping, i.e all KOut/VOut pairs from all
executed MapCombineCommands will be collocated on a node N where KOut is
hashed to and value for KOut will be a list of all VOut values. We
essentially collect all VOut values under each KOut for all executed
MapCombineCommands.
At this point MapCombineCommand has finished its execution; list of KOut
keys is returned to a master node and its MapReduceTask. We do not
return VOut values as we do not need them at master task node.
MapReduceTask is ready to start with reduce phase.
Reduce phase
MapReduceTask initializes ReduceCommand with a user specified Reducer.
For each key KOut collected from a map phase we group them by execution
node N they are hashed to. For each node N and its grouped input KOut
keys MapReduceTask creates a ReduceCommand and sends it to a node N
where KOut keys are hashed. Once loaded on target execution node,
ReduceCommand for each KOut key grabs list of values VOut and invokes:
VOut reduce(KOut reducedKey, Iterator<VOut> iter).
A result of ReduceCommand is a map M where each key is KOut and value is
VOut. Each Infinispan execution node N returns one map M where each key
KOut is hashed to N and each VOut is KOut's reduced value.
When all ReduceCommands return to a calling node, MapReduceTask simply
combines all these M maps and returns final Map<KOut, VOut> as a result
of MapReduceTask. All intermediate KOut->List<VOut> maps left on
Infinispan cluster are then cleaned up.
This all makes sense as well, however one problem with the consistent hash approach is
that it is prone to change when there is a topology change. How would you deal with that?
Would you maintain a history of consistent hashes?
Cheers
Manik
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
8:07 p.m.
On 12-02-20 3:43 PM, Manik Surtani wrote:
I was under the impression reduce is distributed too? Don't we
do the mapping on each node, then a first-pass reduce on each node too, before streaming
results back to the caller node?
What we do in first-pass reduce is essentially
combine and we should not
do that blindly because this eager reduction/combine only works when
reduce function is both /commutative/ and /associative/! This can lead
to problems when it is not:
http://philippeadjiman.com/blog/2010/01/14/hadoop-tutorial-series-issue-4...
So yes first-pass reduce is distributed but second-phase reduce should
be distributed as well! Currently it is not!
This all makes sense as well, however one problem with the consistent
hash approach is that it is prone to change when there is a topology change. How would
you deal with that? Would you maintain a history of consistent hashes?
I don't think I understand! Even if there is a topology change
intermediate results Map<KOut, List<VOut>> will be migrated, all we need
is KOut's, we can hash it and find out where List<VOut> are, no?
Vladimir
8:15 p.m.
On 20 Feb 2012, at 19:07, Vladimir Blagojevic wrote:
On 12-02-20 3:43 PM, Manik Surtani wrote:
>
> I was under the impression reduce is distributed too? Don't we do the mapping on
each node, then a first-pass reduce on each node too, before streaming results back to the
caller node?
What we do in first-pass reduce is essentially combine and we should not do that blindly
because this eager reduction/combine only works when reduce function is both commutative
and associative! This can lead to problems when it is not:
http://philippeadjiman.com/blog/2010/01/14/hadoop-tutorial-series-issue-4...
So yes first-pass reduce is distributed but second-phase reduce should be distributed as
well! Currently it is not!
Ok.
> This all makes sense as well, however one problem with the
consistent hash approach is that it is prone to change when there is a topology change.
How would you deal with that? Would you maintain a history of consistent hashes?
>
>
I don't think I understand! Even if there is a topology change intermediate results
Map<KOut, List<VOut>> will be migrated, all we need is KOut's, we can hash
it and find out where List<VOut> are, no?
Well, if you assign a set of tasks to specific nodes based on a consistent hash, the
topology then changes, you'd lose the information on where you sent specific tasks.
Cheers
Manik
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
9:20 p.m.
On 12-02-20 4:15 PM, Manik Surtani wrote:
Well, if you assign a set of tasks to specific nodes based on a
consistent hash, the topology then changes, you'd lose the information
on where you sent specific tasks.
True, but do we need to care about that? If a task fails due to node
going down we just need to detect such occurrence and migrate a task to
a back up node!
11:50 a.m.
On 20 Feb 2012, at 20:20, Vladimir Blagojevic wrote:
On 12-02-20 4:15 PM, Manik Surtani wrote:
>
> Well, if you assign a set of tasks to specific nodes based on a consistent hash, the
topology then changes, you'd lose the information on where you sent specific tasks.
>
True, but do we need to care about that? If a task fails due to node going down we just
need to detect such occurrence and migrate a task to a back up node!
Yes. As long as you have taken this into account in your design.
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
12:38 p.m.
Hi Vladimir,
I'm a bit late, sorry. My comments below:
On 14 Feb 2012, at 15:19, Vladimir Blagojevic wrote:
Hey guys,
Before moving forward with next iteration of map reduce I wanted to hear
your thoughts about the following proposal. After we agree on the
general direction I will transcribe the agreed design on a wiki page and
start implementation.
Shortcoming of current map reduce implementation
While our current map reduce implementation is more than a proof of a
concept there are several drawbacks preventing it from being an
industrial grade map reduce solution. The main drawback is the inability
of the current solution to deal with a large data (in GB/TB) map reduce
problems. This shortcoming is mainly around our reduce phase execution.
Reduce phase, as you might know, is currently done on a single
Infinispan master task node; reduce phase of map reduce problems we can
support (data size wise) is therefore shrunk to a working memory of a
single node.
Proposed solution
The proposed solution involves distributing execution of reduce phase
tasks across the cluster thus effectively achieving higher reduce task
parallelization and at the same time removing the above mentioned reduce
phase restriction.
Something like a hierarchy (tree) of reducers?
Effectively leveraging our consistent hashing
solution even further we can parallelize reduce phase and elevate our
map reduce solution to an industrial level. Here is how we can achieve that.
Map phase
MapReduceTask, as it currently does, will hash task input keys and group
them by execution node N they are hashed to. For each node N and its
grouped input KIn keys MapReduceTask creates a MapCombineCommand which
is migrated to an execution target node N. MapCombineCommand is similar
to current MapReduceCommand. MapCombineCommand takes an instance of a
Mapper and an instance of a Reducer, which is a combiner [1].
Feels like our
Reducer is more of an Combiner as described by [1], as it enforce the result of the reduce
(KOut) to have the same type as the entries it consumes (KOut).
Don't you think that if we add the Combiner/MapCombineCommand we can loosen up this
constraint and allow the Reducer to output something else than KOut?
Once loaded into target execution node MapCombineCommand takes each
local KIn key and executes Mapper method void map(KIn key, VIn value,
Collector<KOut, VOut> collector). Results are collected to a common
Collector<KOut, VOut> collector and combine phase is initiated. A
Combiner, if specified, takes KOut keys and imediatelly invokes reduce
phase on keys. The result of mapping phase executed on each node is
<KOut, VOut> map M. There will be one resulting M map per execution node N.
So the combiner has the same output as the Mapper - just that this time it is in the
form of an Map<KOut, VOut> instead of a Collector<Kout,Vout>?
Which is okay IMO, as at the end of the day the Combiner is just an extra polishing added
to the mapping...
At the end of combine phase instead of returning map M to the master
task node (as we currently do), we now hash each KOut in map M and group
KOut keys by the execution node N they are hashed to. Each group of KOut
keys and its VOut values, hashed to the same node, is wrapped with a new
command Migrate. Command Migrate, which is very similar to
PutKeyValueCommand,executed on Infinispan target node N esentially
maintains KOut K -> List<VOut> mapping, i.e all KOut/VOut pairs from all
executed MapCombineCommands will be collocated on a node N where KOut is
hashed to and value for KOut will be a list of all VOut values. We
essentially collect all VOut values under each KOut for all executed
MapCombineCommands.
This potentially involves each node sending some state to all other nodes, i.e.
(clusterSize-1)^2 RPCs. Not that this is necessarily a bad thing, as it is there to reduce
contention on a single node.
At this point MapCombineCommand has finished its execution; list of
KOut
keys is returned to a master node and its MapReduceTask. We do not
return VOut values as we do not need them at master task node.
MapReduceTask is ready to start with reduce phase.
If you plan to add this to a design document, I think it would be a good idea to
show an example of how the combiner would be used with the word numbering algorithm. You
explained this quite well here, but that would make sure we're all on the same page on
the role of combiner :-)
Reduce phase
MapReduceTask initializes ReduceCommand with a user specified Reducer.
For each key KOut collected from a map phase we group them by execution
node N they are hashed to. For each node N and its grouped input KOut
keys MapReduceTask creates a ReduceCommand and sends it to a node N
where KOut keys are hashed.
the originator (node on which the M/R is run), at this
point, doesn't know the entire set of KOut values produces, so it can't determine
the list of nodes where to send the ReduceCommand - it would have to send it to all nodes,
right?
Once loaded on target execution node,
ReduceCommand for each KOut key grabs list of values VOut and invokes:
VOut reduce(KOut reducedKey, Iterator<VOut> iter).
A result of ReduceCommand is a map M where each key is KOut and value is
VOut. Each Infinispan execution node N returns one map M where each key
KOut is hashed to N and each VOut is KOut's reduced value.
When all ReduceCommands return to a calling node, MapReduceTask simply
combines all these M maps and returns final Map<KOut, VOut> as a result
of MapReduceTask. All intermediate KOut->List<VOut> maps left on
Infinispan cluster are then cleaned up.
[1] See section 4.3 of http://research.google.com/archive/mapreduce.html
7:48 p.m.
Hey Mircea,
On 12-03-20 7:38 AM, Mircea Markus wrote:
Hi Vladimir,
I'm a bit late, sorry. My comments below:
Feels like our Reducer is more of an Combiner as described by [1], as it enforce the
result of the reduce (KOut) to have the same type as the entries it consumes (KOut).
Don't you think that if we add the Combiner/MapCombineCommand we can loosen up this
constraint and allow the Reducer to output something else than KOut?
I tried playing
with this but could not hash out nice API interface that
did not complicate everything too much. Let me see your API interface
proposal Mircea.
> Once loaded into target execution node MapCombineCommand takes each
> local KIn key and executes Mapper method void map(KIn key, VIn value,
> Collector<KOut, VOut> collector). Results are collected to a common
> Collector<KOut, VOut> collector and combine phase is initiated. A
> Combiner, if specified, takes KOut keys and imediatelly invokes reduce
> phase on keys. The result of mapping phase executed on each node is
> <KOut, VOut> map M. There will be one resulting M map per execution node N.
So the combiner has the same output as the Mapper - just that this time it is in the form
of an Map<KOut, VOut> instead of a Collector<Kout,Vout>?
Which is okay IMO, as at the end of the day the Combiner is just an extra polishing added
to the mapping...
Yes exactly. Combiner is just a polisher allowing less reduction later
on - however one can only use// combiner only if reduce function is both
associative and commutative, see :
http://philippeadjiman.com/blog/2010/01/14/hadoop-tutorial-series-issue-4...
> At the end of combine phase instead of returning map M to the
master
> task node (as we currently do), we now hash each KOut in map M and group
> KOut keys by the execution node N they are hashed to. Each group of KOut
> keys and its VOut values, hashed to the same node, is wrapped with a new
> command Migrate. Command Migrate, which is very similar to
> PutKeyValueCommand,executed on Infinispan target node N esentially
> maintains KOut K -> List<VOut> mapping, i.e all KOut/VOut pairs from all
> executed MapCombineCommands will be collocated on a node N where KOut is
> hashed to and value for KOut will be a list of all VOut values. We
> essentially collect all VOut values under each KOut for all executed
> MapCombineCommands.
This potentially involves each node sending some state to all other nodes, i.e.
(clusterSize-1)^2 RPCs. Not that this is necessarily a bad thing, as it is there to reduce
contention on a single node.
Each node sending state to all other nodes, really? Isn't it that each
node sends state to numOwners nodes?
> At this point MapCombineCommand has finished its execution; list of KOut
> keys is returned to a master node and its MapReduceTask. We do not
> return VOut values as we do not need them at master task node.
> MapReduceTask is ready to start with reduce phase.
>
If you plan to add this to a design document, I think it would be a good idea to show an
example of how the combiner would be used with the word numbering algorithm. You explained
this quite well here, but that would make sure we're all on the same page on the role
of combiner :-)
Deal!
> Reduce phase
>
>
> MapReduceTask initializes ReduceCommand with a user specified Reducer.
> For each key KOut collected from a map phase we group them by execution
> node N they are hashed to. For each node N and its grouped input KOut
> keys MapReduceTask creates a ReduceCommand and sends it to a node N
> where KOut keys are hashed.
the originator (node on which the M/R is run), at this point, doesn't know the entire
set of KOut values produces, so it can't determine the list of nodes where to send the
ReduceCommand - it would have to send it to all nodes, right?
No, it does know all KOut values but it does not know know all VOut
values (they are on the grid)!
Thanks a lot Mircea! Do you think overall this design makes sense? Are
you concerned about level of traffic for migrate intermediate results phase?
Regards,
Vladimir
8:39 p.m.
Thanks everyone for the feedback. I will maintain the
design/requirements document at
https://community.jboss.org/wiki/Infinispan60-MapReduceEnhancements
If anything remains unclear do tell me. Keep the feedback coming.
Regards,
Vladimir
P.S I will also create another document that is related to enhancements
to both distributed executors and map reduce - mainly fail-over,
topology awareness, task interruption/cancellation etc
On 12-02-14 10:19 AM, Vladimir Blagojevic wrote:
Hey guys,
Before moving forward with next iteration of map reduce I wanted to hear
your thoughts about the following proposal. After we agree on the
general direction I will transcribe the agreed design on a wiki page and
start implementation.
Shortcoming of current map reduce implementation
While our current map reduce implementation is more than a proof of a
concept there are several drawbacks preventing it from being an
industrial grade map reduce solution. The main drawback is the inability
of the current solution to deal with a large data (in GB/TB) map reduce
problems. This shortcoming is mainly around our reduce phase execution.
Reduce phase, as you might know, is currently done on a single
Infinispan master task node; reduce phase of map reduce problems we can
support (data size wise) is therefore shrunk to a working memory of a
single node.
Proposed solution
The proposed solution involves distributing execution of reduce phase
tasks across the cluster thus effectively achieving higher reduce task
parallelization and at the same time removing the above mentioned reduce
phase restriction. Effectively leveraging our consistent hashing
solution even further we can parallelize reduce phase and elevate our
map reduce solution to an industrial level. Here is how we can achieve that.
Map phase
MapReduceTask, as it currently does, will hash task input keys and group
them by execution node N they are hashed to. For each node N and its
grouped input KIn keys MapReduceTask creates a MapCombineCommand which
is migrated to an execution target node N. MapCombineCommand is similar
to current MapReduceCommand. MapCombineCommand takes an instance of a
Mapper and an instance of a Reducer, which is a combiner [1].
Once loaded into target execution node MapCombineCommand takes each
local KIn key and executes Mapper method void map(KIn key, VIn value,
Collector<KOut, VOut> collector). Results are collected to a common
Collector<KOut, VOut> collector and combine phase is initiated. A
Combiner, if specified, takes KOut keys and imediatelly invokes reduce
phase on keys. The result of mapping phase executed on each node is
<KOut, VOut> map M. There will be one resulting M map per execution node N.
At the end of combine phase instead of returning map M to the master
task node (as we currently do), we now hash each KOut in map M and group
KOut keys by the execution node N they are hashed to. Each group of KOut
keys and its VOut values, hashed to the same node, is wrapped with a new
command Migrate. Command Migrate, which is very similar to
PutKeyValueCommand,executed on Infinispan target node N esentially
maintains KOut K -> List<VOut> mapping, i.e all KOut/VOut pairs from all
executed MapCombineCommands will be collocated on a node N where KOut is
hashed to and value for KOut will be a list of all VOut values. We
essentially collect all VOut values under each KOut for all executed
MapCombineCommands.
At this point MapCombineCommand has finished its execution; list of KOut
keys is returned to a master node and its MapReduceTask. We do not
return VOut values as we do not need them at master task node.
MapReduceTask is ready to start with reduce phase.
Reduce phase
MapReduceTask initializes ReduceCommand with a user specified Reducer.
For each key KOut collected from a map phase we group them by execution
node N they are hashed to. For each node N and its grouped input KOut
keys MapReduceTask creates a ReduceCommand and sends it to a node N
where KOut keys are hashed. Once loaded on target execution node,
ReduceCommand for each KOut key grabs list of values VOut and invokes:
VOut reduce(KOut reducedKey, Iterator<VOut> iter).
A result of ReduceCommand is a map M where each key is KOut and value is
VOut. Each Infinispan execution node N returns one map M where each key
KOut is hashed to N and each VOut is KOut's reduced value.
When all ReduceCommands return to a calling node, MapReduceTask simply
combines all these M maps and returns final Map<KOut, VOut> as a result
of MapReduceTask. All intermediate KOut->List<VOut> maps left on
Infinispan cluster are then cleaned up.
[1] See section 4.3 of http://research.google.com/archive/mapreduce.html
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
4:06 p.m.
Some thoughts:
1. Failover policy.
* Does this rely on being run in a CDI environment? (RandomNodeTaskFailoverPolicy has a
DistributedExecutorService @Injected). If so, then -1. While it *should work* in a CDI
environment, it shouldn't *require* CDI.
* Do you plan to include any canned failover policies? If so, what are these?
* Shouldn't DistributedTaskFailoverPolicy be parameterised to return the same type as
DistributedFuture.get() rather than an untyped Object?
2. Task mapping policy
* So the existing policies of "execute everywhere" and "execute on data
owner" (and possibly a new one, "execute on PRIMARY data owner") would all
implement this policy interface?
Cheers
Manik
On 1 May 2012, at 19:39, Vladimir Blagojevic wrote:
Thanks everyone for the feedback. I will maintain the
design/requirements document at
https://community.jboss.org/wiki/Infinispan60-MapReduceEnhancements
If anything remains unclear do tell me. Keep the feedback coming.
Regards,
Vladimir
P.S I will also create another document that is related to enhancements
to both distributed executors and map reduce - mainly fail-over,
topology awareness, task interruption/cancellation etc
On 12-02-14 10:19 AM, Vladimir Blagojevic wrote:
> Hey guys,
>
> Before moving forward with next iteration of map reduce I wanted to hear
> your thoughts about the following proposal. After we agree on the
> general direction I will transcribe the agreed design on a wiki page and
> start implementation.
>
>
> Shortcoming of current map reduce implementation
>
> While our current map reduce implementation is more than a proof of a
> concept there are several drawbacks preventing it from being an
> industrial grade map reduce solution. The main drawback is the inability
> of the current solution to deal with a large data (in GB/TB) map reduce
> problems. This shortcoming is mainly around our reduce phase execution.
> Reduce phase, as you might know, is currently done on a single
> Infinispan master task node; reduce phase of map reduce problems we can
> support (data size wise) is therefore shrunk to a working memory of a
> single node.
>
>
> Proposed solution
>
> The proposed solution involves distributing execution of reduce phase
> tasks across the cluster thus effectively achieving higher reduce task
> parallelization and at the same time removing the above mentioned reduce
> phase restriction. Effectively leveraging our consistent hashing
> solution even further we can parallelize reduce phase and elevate our
> map reduce solution to an industrial level. Here is how we can achieve that.
>
> Map phase
>
> MapReduceTask, as it currently does, will hash task input keys and group
> them by execution node N they are hashed to. For each node N and its
> grouped input KIn keys MapReduceTask creates a MapCombineCommand which
> is migrated to an execution target node N. MapCombineCommand is similar
> to current MapReduceCommand. MapCombineCommand takes an instance of a
> Mapper and an instance of a Reducer, which is a combiner [1].
>
> Once loaded into target execution node MapCombineCommand takes each
> local KIn key and executes Mapper method void map(KIn key, VIn value,
> Collector<KOut, VOut> collector). Results are collected to a common
> Collector<KOut, VOut> collector and combine phase is initiated. A
> Combiner, if specified, takes KOut keys and imediatelly invokes reduce
> phase on keys. The result of mapping phase executed on each node is
> <KOut, VOut> map M. There will be one resulting M map per execution node N.
>
> At the end of combine phase instead of returning map M to the master
> task node (as we currently do), we now hash each KOut in map M and group
> KOut keys by the execution node N they are hashed to. Each group of KOut
> keys and its VOut values, hashed to the same node, is wrapped with a new
> command Migrate. Command Migrate, which is very similar to
> PutKeyValueCommand,executed on Infinispan target node N esentially
> maintains KOut K -> List<VOut> mapping, i.e all KOut/VOut pairs from all
> executed MapCombineCommands will be collocated on a node N where KOut is
> hashed to and value for KOut will be a list of all VOut values. We
> essentially collect all VOut values under each KOut for all executed
> MapCombineCommands.
>
>
> At this point MapCombineCommand has finished its execution; list of KOut
> keys is returned to a master node and its MapReduceTask. We do not
> return VOut values as we do not need them at master task node.
> MapReduceTask is ready to start with reduce phase.
>
>
> Reduce phase
>
>
> MapReduceTask initializes ReduceCommand with a user specified Reducer.
> For each key KOut collected from a map phase we group them by execution
> node N they are hashed to. For each node N and its grouped input KOut
> keys MapReduceTask creates a ReduceCommand and sends it to a node N
> where KOut keys are hashed. Once loaded on target execution node,
> ReduceCommand for each KOut key grabs list of values VOut and invokes:
> VOut reduce(KOut reducedKey, Iterator<VOut> iter).
>
> A result of ReduceCommand is a map M where each key is KOut and value is
> VOut. Each Infinispan execution node N returns one map M where each key
> KOut is hashed to N and each VOut is KOut's reduced value.
>
> When all ReduceCommands return to a calling node, MapReduceTask simply
> combines all these M maps and returns final Map<KOut, VOut> as a result
> of MapReduceTask. All intermediate KOut->List<VOut> maps left on
> Infinispan cluster are then cleaned up.
>
>
> [1] See section 4.3 of http://research.google.com/archive/mapreduce.html
> _______________________________________________
> 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
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
4:34 p.m.
On 12-05-31 10:06 AM, Manik Surtani wrote:
Some thoughts:
1. Failover policy.
* Does this rely on being run in a CDI environment? (RandomNodeTaskFailoverPolicy has a
DistributedExecutorService @Injected). If so, then -1. While it *should work* in a CDI
environment, it shouldn't *require* CDI.
I agree. It was more of a wishful
thinking on my part, it will not be
required.
* Do you plan to include any canned failover policies? If so, what are these?
Yes, at least random and never failover policy.
* Shouldn't DistributedTaskFailoverPolicy be parameterised to return the same type
as DistributedFuture.get() rather than an untyped Object?
I would love to make it
but how can we do it? DefaultExecutorService is
not typed and method of DistributedTaskFailoverPolicy is not either.
2. Task mapping policy
* So the existing policies of "execute everywhere" and "execute on data
owner" (and possibly a new one, "execute on PRIMARY data owner") would all
implement this policy interface?
Yes, for sure.
10:49 a.m.
On 31 May 2012, at 15:34, Vladimir Blagojevic wrote:
On 12-05-31 10:06 AM, Manik Surtani wrote:
> Some thoughts:
>
> 1. Failover policy.
>
> * Does this rely on being run in a CDI environment? (RandomNodeTaskFailoverPolicy
has a DistributedExecutorService @Injected). If so, then -1. While it *should work* in a
CDI environment, it shouldn't *require* CDI.
I agree. It was more of a wishful thinking on my part, it will not be required.
We should update the interface then to indicate how you intend to make the ExecutorService
made available to implementations. :)
>
> * Do you plan to include any canned failover policies? If so, what are these?
Yes, at least random and never failover policy.
>
> * Shouldn't DistributedTaskFailoverPolicy be parameterised to return the same
type as DistributedFuture.get() rather than an untyped Object?
I would love to make it but how can we do it? DefaultExecutorService is not typed and
method of DistributedTaskFailoverPolicy is not either.
Can DTFP not be typed? And even our sub-interface of ExecutorService?
>
> 2. Task mapping policy
>
> * So the existing policies of "execute everywhere" and "execute on
data owner" (and possibly a new one, "execute on PRIMARY data owner") would
all implement this policy interface?
>
Yes, for sure.
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
5:21 p.m.
On 12-06-01 4:49 AM, Manik Surtani wrote:
We should update the interface then to indicate how you intend to
make the ExecutorService made available to implementations. :)
We can provide setDistributedExecutorService(DistributedExecutorService
service) method on DistributedTaskFailoverPolicy interface? However,
users can rather easily construct their own impl of
DistributedTaskFailoverPolicy by passing instance of readily available
DistributedExecutorService through a DistributedTaskFailoverPolicy
constructor for example. The downside is that only programmatic
configuration would work in this case. If we have set method contract
then we can setup custom task failover through xml configuration. WDYT?
>> * Do you plan to include any canned failover policies? If
so, what are these?
> Yes, at least random and never failover policy.
>> * Shouldn't DistributedTaskFailoverPolicy be parameterised to return the
same type as DistributedFuture.get() rather than an untyped Object?
> I would love to make it but how can we do it? DefaultExecutorService is not typed and
method of DistributedTaskFailoverPolicy is not either.
Can DTFP not be typed? And even our sub-interface of ExecutorService?
I am not sure if we should type our sub interface of ExecutorService.
Although I do not know how to argue against it, all sub interfaces and
impls I have seen use generic methods. Probably because generic methods
give you more flexibility. So yes DTFP can by typed albeit not a very
useful thing as DTFP should deal with generic type T rather than
particular type. I think there is no need to type the interface when we
can type method in DTFP. Also up until today I thought that generic
method can infer its parameter type from its argument type only.
However, even this works:
<T> T foo(){...}
Pretty cool!
Regards,
Vladimir
12:48 p.m.
On 1 Jun 2012, at 16:21, Vladimir Blagojevic wrote:
On 12-06-01 4:49 AM, Manik Surtani wrote:
>
> We should update the interface then to indicate how you intend to make the
ExecutorService made available to implementations. :)
We can provide setDistributedExecutorService(DistributedExecutorService service) method
on DistributedTaskFailoverPolicy interface? However, users can rather easily construct
their own impl of DistributedTaskFailoverPolicy by passing instance of readily available
DistributedExecutorService through a DistributedTaskFailoverPolicy constructor for
example. The downside is that only programmatic configuration would work in this case. If
we have set method contract then we can setup custom task failover through xml
configuration. WDYT?
+1 to having a policy that can be configured via XML.
>>> * Do you plan to include any canned failover policies?
If so, what are these?
>> Yes, at least random and never failover policy.
>>> * Shouldn't DistributedTaskFailoverPolicy be parameterised to return the
same type as DistributedFuture.get() rather than an untyped Object?
>> I would love to make it but how can we do it? DefaultExecutorService is not typed
and method of DistributedTaskFailoverPolicy is not either.
> Can DTFP not be typed? And even our sub-interface of ExecutorService?
>
I am not sure if we should type our sub interface of ExecutorService. Although I do not
know how to argue against it, all sub interfaces and impls I have seen use generic
methods. Probably because generic methods give you more flexibility. So yes DTFP can by
typed albeit not a very useful thing as DTFP should deal with generic type T rather than
particular type. I think there is no need to type the interface when we can type method in
DTFP. Also up until today I thought that generic method can infer its parameter type from
its argument type only. However, even this works:
<T> T foo(){...}
Pretty cool!
Do you have an example of what you are proposing by inferring its type?
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
4:29 p.m.
On 12-06-06 6:48 AM, Manik Surtani wrote:
+1 to having a policy that can be configured via XML.
Done!
Do you have an example of what you are proposing by inferring its type?
The latest proposal is
https://community.jboss.org/wiki/Infinispan60-DistributedExecutionEnhance...
See DistributedTaskFailoverPolicy executionFailed method.
Regards,
Vladimir
4:47 p.m.
Cool. Are you working on this now? How long do you reckon before we can unleash it on
the public? ;)
On 8 Jun 2012, at 15:29, Vladimir Blagojevic wrote:
On 12-06-06 6:48 AM, Manik Surtani wrote:
>
>
> +1 to having a policy that can be configured via XML.
Done!
>
> Do you have an example of what you are proposing by inferring its type?
>
The latest proposal is
https://community.jboss.org/wiki/Infinispan60-DistributedExecutionEnhance...
See DistributedTaskFailoverPolicy executionFailed method.
Regards,
Vladimir
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
5:17 p.m.
On 12-06-08 10:47 AM, Manik Surtani wrote:
Cool. Are you working on this now? How long do you reckon before we
can unleash it on the public? ;)
Well I wanted to start early, gather feedback in a few iterations and
have it ready for 6.0. Currently, I am half way through implementing
Map/Reduce new design. I'll post that branch and will get back to this
one then. We need to somehow raise awareness about these changes and
Galder suggested that I blog with concrete examples, pitch for feedback,
you guys tweet and so on....
12:28 p.m.
Does it affect the current API in any way? If not, then I suggest putting it in 5.2.0 if
it is ready. Blogging about it is cool, but you won't get any real feedback unless
people can actually use it.
On 8 Jun 2012, at 16:17, Vladimir Blagojevic wrote:
On 12-06-08 10:47 AM, Manik Surtani wrote:
> Cool. Are you working on this now? How long do you reckon before we can unleash it
on the public? ;)
>
>
Well I wanted to start early, gather feedback in a few iterations and have it ready for
6.0. Currently, I am half way through implementing Map/Reduce new design. I'll post
that branch and will get back to this one then. We need to somehow raise awareness about
these changes and Galder suggested that I blog with concrete examples, pitch for feedback,
you guys tweet and so on....
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
4345
days inactive
4461
days old
infinispan-dev@lists.jboss.org
17 comments
3 participants
participants (3)
-
Manik Surtani -
Mircea Markus -
Vladimir Blagojevic