Jim Twensky
2012-10-05, 16:31
Harsh J
2012-10-05, 17:18
Jim Twensky
2012-10-05, 17:43
Harsh J
2012-10-05, 17:54
Jim Twensky
2012-10-05, 18:02
Bertrand Dechoux
2012-10-08, 10:39
Fabio Pitzolu
2012-10-08, 10:44
Bertrand Dechoux
2012-10-08, 10:51
Jim Twensky
2012-10-08, 19:09
Michael Segel
2012-10-08, 19:19
-
Chaning Multiple Reducers: Reduce -> Reduce -> ReduceJim Twensky 2012-10-05, 16:31
Hi,
I have a complex Hadoop job that iterates over large graph data
multiple times until some convergence condition is met. I know that
the map output goes to the local disk of each particular mapper first,
and then fetched by the reducers before the reduce tasks start. I can
see that this is an overhead, and it theory we can ship the data
directly from mappers to reducers, without serializing on the local
disk first. I understand that this step is necessary for fault
tolerance and it is an essential building block of MapReduce.
In my application, the map process consists of identity mappers which
read the input from HDFS and ship it to reducers. Essentially, what I
am doing is applying chains of reduce jobs until the algorithm
converges. My question is, can I bypass the serialization of the local
data and ship it from mappers to reducers immediately (as soon as I
call context.write() in my mapper class)? If not, are there any other
MR platforms that can do this? I've been searching around and couldn't
see anything similar to what I need. Hadoop On Line is a prototype and
has some similar functionality but it hasn't been updated for a while.
Note: I know about ChainMapper and ChainReducer classes but I don't
want to chain multiple mappers in the same local node. I want to chain
multiple reduce functions globally so the data flow looks like: Map ->
Reduce -> Reduce -> Reduce, which means each reduce operation is
followed by a shuffle and sort essentially bypassing the map
operation.
I have a complex Hadoop job that iterates over large graph data
multiple times until some convergence condition is met. I know that
the map output goes to the local disk of each particular mapper first,
and then fetched by the reducers before the reduce tasks start. I can
see that this is an overhead, and it theory we can ship the data
directly from mappers to reducers, without serializing on the local
disk first. I understand that this step is necessary for fault
tolerance and it is an essential building block of MapReduce.
In my application, the map process consists of identity mappers which
read the input from HDFS and ship it to reducers. Essentially, what I
am doing is applying chains of reduce jobs until the algorithm
converges. My question is, can I bypass the serialization of the local
data and ship it from mappers to reducers immediately (as soon as I
call context.write() in my mapper class)? If not, are there any other
MR platforms that can do this? I've been searching around and couldn't
see anything similar to what I need. Hadoop On Line is a prototype and
has some similar functionality but it hasn't been updated for a while.
Note: I know about ChainMapper and ChainReducer classes but I don't
want to chain multiple mappers in the same local node. I want to chain
multiple reduce functions globally so the data flow looks like: Map ->
Reduce -> Reduce -> Reduce, which means each reduce operation is
followed by a shuffle and sort essentially bypassing the map
operation.
-
Re: Chaning Multiple Reducers: Reduce -> Reduce -> ReduceHarsh J 2012-10-05, 17:18
Hey Jim,
Are you looking to re-sort or re-partition your data by a different
key or key combo after each output from reduce?
On Fri, Oct 5, 2012 at 10:01 PM, Jim Twensky <[EMAIL PROTECTED]> wrote:
> Hi,
>
> I have a complex Hadoop job that iterates over large graph data
> multiple times until some convergence condition is met. I know that
> the map output goes to the local disk of each particular mapper first,
> and then fetched by the reducers before the reduce tasks start. I can
> see that this is an overhead, and it theory we can ship the data
> directly from mappers to reducers, without serializing on the local
> disk first. I understand that this step is necessary for fault
> tolerance and it is an essential building block of MapReduce.
>
> In my application, the map process consists of identity mappers which
> read the input from HDFS and ship it to reducers. Essentially, what I
> am doing is applying chains of reduce jobs until the algorithm
> converges. My question is, can I bypass the serialization of the local
> data and ship it from mappers to reducers immediately (as soon as I
> call context.write() in my mapper class)? If not, are there any other
> MR platforms that can do this? I've been searching around and couldn't
> see anything similar to what I need. Hadoop On Line is a prototype and
> has some similar functionality but it hasn't been updated for a while.
>
> Note: I know about ChainMapper and ChainReducer classes but I don't
> want to chain multiple mappers in the same local node. I want to chain
> multiple reduce functions globally so the data flow looks like: Map ->
> Reduce -> Reduce -> Reduce, which means each reduce operation is
> followed by a shuffle and sort essentially bypassing the map
> operation.
--
Harsh J
Are you looking to re-sort or re-partition your data by a different
key or key combo after each output from reduce?
On Fri, Oct 5, 2012 at 10:01 PM, Jim Twensky <[EMAIL PROTECTED]> wrote:
> Hi,
>
> I have a complex Hadoop job that iterates over large graph data
> multiple times until some convergence condition is met. I know that
> the map output goes to the local disk of each particular mapper first,
> and then fetched by the reducers before the reduce tasks start. I can
> see that this is an overhead, and it theory we can ship the data
> directly from mappers to reducers, without serializing on the local
> disk first. I understand that this step is necessary for fault
> tolerance and it is an essential building block of MapReduce.
>
> In my application, the map process consists of identity mappers which
> read the input from HDFS and ship it to reducers. Essentially, what I
> am doing is applying chains of reduce jobs until the algorithm
> converges. My question is, can I bypass the serialization of the local
> data and ship it from mappers to reducers immediately (as soon as I
> call context.write() in my mapper class)? If not, are there any other
> MR platforms that can do this? I've been searching around and couldn't
> see anything similar to what I need. Hadoop On Line is a prototype and
> has some similar functionality but it hasn't been updated for a while.
>
> Note: I know about ChainMapper and ChainReducer classes but I don't
> want to chain multiple mappers in the same local node. I want to chain
> multiple reduce functions globally so the data flow looks like: Map ->
> Reduce -> Reduce -> Reduce, which means each reduce operation is
> followed by a shuffle and sort essentially bypassing the map
> operation.
--
Harsh J
-
Re: Chaning Multiple Reducers: Reduce -> Reduce -> ReduceJim Twensky 2012-10-05, 17:43
Hi Harsh,
Yes, there is actually a "hidden" map stage, that generates new
<key,value> pairs based on the last reduce output but I can create
those records during the reduce step instead and get rid of the
intermediate map computation completely. The idea is to apply the map
function to each output of the reduce inside the reduce class and emit
the result as the output of the reducer.
Jim
On Fri, Oct 5, 2012 at 12:18 PM, Harsh J <[EMAIL PROTECTED]> wrote:
> Hey Jim,
>
> Are you looking to re-sort or re-partition your data by a different
> key or key combo after each output from reduce?
>
> On Fri, Oct 5, 2012 at 10:01 PM, Jim Twensky <[EMAIL PROTECTED]> wrote:
>> Hi,
>>
>> I have a complex Hadoop job that iterates over large graph data
>> multiple times until some convergence condition is met. I know that
>> the map output goes to the local disk of each particular mapper first,
>> and then fetched by the reducers before the reduce tasks start. I can
>> see that this is an overhead, and it theory we can ship the data
>> directly from mappers to reducers, without serializing on the local
>> disk first. I understand that this step is necessary for fault
>> tolerance and it is an essential building block of MapReduce.
>>
>> In my application, the map process consists of identity mappers which
>> read the input from HDFS and ship it to reducers. Essentially, what I
>> am doing is applying chains of reduce jobs until the algorithm
>> converges. My question is, can I bypass the serialization of the local
>> data and ship it from mappers to reducers immediately (as soon as I
>> call context.write() in my mapper class)? If not, are there any other
>> MR platforms that can do this? I've been searching around and couldn't
>> see anything similar to what I need. Hadoop On Line is a prototype and
>> has some similar functionality but it hasn't been updated for a while.
>>
>> Note: I know about ChainMapper and ChainReducer classes but I don't
>> want to chain multiple mappers in the same local node. I want to chain
>> multiple reduce functions globally so the data flow looks like: Map ->
>> Reduce -> Reduce -> Reduce, which means each reduce operation is
>> followed by a shuffle and sort essentially bypassing the map
>> operation.
>
>
>
> --
> Harsh J
Yes, there is actually a "hidden" map stage, that generates new
<key,value> pairs based on the last reduce output but I can create
those records during the reduce step instead and get rid of the
intermediate map computation completely. The idea is to apply the map
function to each output of the reduce inside the reduce class and emit
the result as the output of the reducer.
Jim
On Fri, Oct 5, 2012 at 12:18 PM, Harsh J <[EMAIL PROTECTED]> wrote:
> Hey Jim,
>
> Are you looking to re-sort or re-partition your data by a different
> key or key combo after each output from reduce?
>
> On Fri, Oct 5, 2012 at 10:01 PM, Jim Twensky <[EMAIL PROTECTED]> wrote:
>> Hi,
>>
>> I have a complex Hadoop job that iterates over large graph data
>> multiple times until some convergence condition is met. I know that
>> the map output goes to the local disk of each particular mapper first,
>> and then fetched by the reducers before the reduce tasks start. I can
>> see that this is an overhead, and it theory we can ship the data
>> directly from mappers to reducers, without serializing on the local
>> disk first. I understand that this step is necessary for fault
>> tolerance and it is an essential building block of MapReduce.
>>
>> In my application, the map process consists of identity mappers which
>> read the input from HDFS and ship it to reducers. Essentially, what I
>> am doing is applying chains of reduce jobs until the algorithm
>> converges. My question is, can I bypass the serialization of the local
>> data and ship it from mappers to reducers immediately (as soon as I
>> call context.write() in my mapper class)? If not, are there any other
>> MR platforms that can do this? I've been searching around and couldn't
>> see anything similar to what I need. Hadoop On Line is a prototype and
>> has some similar functionality but it hasn't been updated for a while.
>>
>> Note: I know about ChainMapper and ChainReducer classes but I don't
>> want to chain multiple mappers in the same local node. I want to chain
>> multiple reduce functions globally so the data flow looks like: Map ->
>> Reduce -> Reduce -> Reduce, which means each reduce operation is
>> followed by a shuffle and sort essentially bypassing the map
>> operation.
>
>
>
> --
> Harsh J
-
Re: Chaning Multiple Reducers: Reduce -> Reduce -> ReduceHarsh J 2012-10-05, 17:54
Would it then be right to assume that the keys produced by the reduced
partition at one stage would be isolated to its partition alone and
not occur in any of the other partition outputs? I'm guessing not,
based on the nature of your data?
I'm trying to understand why shuffling is good to be avoided here, and
if it can be in some ways, given the data. As I see it, you need
re-sort based on the new key per partition, but not the shuffle? Or am
I wrong?
On Fri, Oct 5, 2012 at 11:13 PM, Jim Twensky <[EMAIL PROTECTED]> wrote:
> Hi Harsh,
>
> Yes, there is actually a "hidden" map stage, that generates new
> <key,value> pairs based on the last reduce output but I can create
> those records during the reduce step instead and get rid of the
> intermediate map computation completely. The idea is to apply the map
> function to each output of the reduce inside the reduce class and emit
> the result as the output of the reducer.
>
> Jim
>
> On Fri, Oct 5, 2012 at 12:18 PM, Harsh J <[EMAIL PROTECTED]> wrote:
>> Hey Jim,
>>
>> Are you looking to re-sort or re-partition your data by a different
>> key or key combo after each output from reduce?
>>
>> On Fri, Oct 5, 2012 at 10:01 PM, Jim Twensky <[EMAIL PROTECTED]> wrote:
>>> Hi,
>>>
>>> I have a complex Hadoop job that iterates over large graph data
>>> multiple times until some convergence condition is met. I know that
>>> the map output goes to the local disk of each particular mapper first,
>>> and then fetched by the reducers before the reduce tasks start. I can
>>> see that this is an overhead, and it theory we can ship the data
>>> directly from mappers to reducers, without serializing on the local
>>> disk first. I understand that this step is necessary for fault
>>> tolerance and it is an essential building block of MapReduce.
>>>
>>> In my application, the map process consists of identity mappers which
>>> read the input from HDFS and ship it to reducers. Essentially, what I
>>> am doing is applying chains of reduce jobs until the algorithm
>>> converges. My question is, can I bypass the serialization of the local
>>> data and ship it from mappers to reducers immediately (as soon as I
>>> call context.write() in my mapper class)? If not, are there any other
>>> MR platforms that can do this? I've been searching around and couldn't
>>> see anything similar to what I need. Hadoop On Line is a prototype and
>>> has some similar functionality but it hasn't been updated for a while.
>>>
>>> Note: I know about ChainMapper and ChainReducer classes but I don't
>>> want to chain multiple mappers in the same local node. I want to chain
>>> multiple reduce functions globally so the data flow looks like: Map ->
>>> Reduce -> Reduce -> Reduce, which means each reduce operation is
>>> followed by a shuffle and sort essentially bypassing the map
>>> operation.
>>
>>
>>
>> --
>> Harsh J
--
Harsh J
partition at one stage would be isolated to its partition alone and
not occur in any of the other partition outputs? I'm guessing not,
based on the nature of your data?
I'm trying to understand why shuffling is good to be avoided here, and
if it can be in some ways, given the data. As I see it, you need
re-sort based on the new key per partition, but not the shuffle? Or am
I wrong?
On Fri, Oct 5, 2012 at 11:13 PM, Jim Twensky <[EMAIL PROTECTED]> wrote:
> Hi Harsh,
>
> Yes, there is actually a "hidden" map stage, that generates new
> <key,value> pairs based on the last reduce output but I can create
> those records during the reduce step instead and get rid of the
> intermediate map computation completely. The idea is to apply the map
> function to each output of the reduce inside the reduce class and emit
> the result as the output of the reducer.
>
> Jim
>
> On Fri, Oct 5, 2012 at 12:18 PM, Harsh J <[EMAIL PROTECTED]> wrote:
>> Hey Jim,
>>
>> Are you looking to re-sort or re-partition your data by a different
>> key or key combo after each output from reduce?
>>
>> On Fri, Oct 5, 2012 at 10:01 PM, Jim Twensky <[EMAIL PROTECTED]> wrote:
>>> Hi,
>>>
>>> I have a complex Hadoop job that iterates over large graph data
>>> multiple times until some convergence condition is met. I know that
>>> the map output goes to the local disk of each particular mapper first,
>>> and then fetched by the reducers before the reduce tasks start. I can
>>> see that this is an overhead, and it theory we can ship the data
>>> directly from mappers to reducers, without serializing on the local
>>> disk first. I understand that this step is necessary for fault
>>> tolerance and it is an essential building block of MapReduce.
>>>
>>> In my application, the map process consists of identity mappers which
>>> read the input from HDFS and ship it to reducers. Essentially, what I
>>> am doing is applying chains of reduce jobs until the algorithm
>>> converges. My question is, can I bypass the serialization of the local
>>> data and ship it from mappers to reducers immediately (as soon as I
>>> call context.write() in my mapper class)? If not, are there any other
>>> MR platforms that can do this? I've been searching around and couldn't
>>> see anything similar to what I need. Hadoop On Line is a prototype and
>>> has some similar functionality but it hasn't been updated for a while.
>>>
>>> Note: I know about ChainMapper and ChainReducer classes but I don't
>>> want to chain multiple mappers in the same local node. I want to chain
>>> multiple reduce functions globally so the data flow looks like: Map ->
>>> Reduce -> Reduce -> Reduce, which means each reduce operation is
>>> followed by a shuffle and sort essentially bypassing the map
>>> operation.
>>
>>
>>
>> --
>> Harsh J
--
Harsh J
-
Re: Chaning Multiple Reducers: Reduce -> Reduce -> ReduceJim Twensky 2012-10-05, 18:02
Hi Harsh,
The hidden map operation which is applied to the reduced partition at
one stage can generate keys that are outside of the range covered by
that particular reducer. I still need to have the many-to-many
communication from reduce step k to reduce step k+1. Otherwise, I
think the ChainReducer would do the job and apply multiple maps to
each isolated partition produced by the reducer.
Jim
On Fri, Oct 5, 2012 at 12:54 PM, Harsh J <[EMAIL PROTECTED]> wrote:
> Would it then be right to assume that the keys produced by the reduced
> partition at one stage would be isolated to its partition alone and
> not occur in any of the other partition outputs? I'm guessing not,
> based on the nature of your data?
>
> I'm trying to understand why shuffling is good to be avoided here, and
> if it can be in some ways, given the data. As I see it, you need
> re-sort based on the new key per partition, but not the shuffle? Or am
> I wrong?
>
> On Fri, Oct 5, 2012 at 11:13 PM, Jim Twensky <[EMAIL PROTECTED]> wrote:
>> Hi Harsh,
>>
>> Yes, there is actually a "hidden" map stage, that generates new
>> <key,value> pairs based on the last reduce output but I can create
>> those records during the reduce step instead and get rid of the
>> intermediate map computation completely. The idea is to apply the map
>> function to each output of the reduce inside the reduce class and emit
>> the result as the output of the reducer.
>>
>> Jim
>>
>> On Fri, Oct 5, 2012 at 12:18 PM, Harsh J <[EMAIL PROTECTED]> wrote:
>>> Hey Jim,
>>>
>>> Are you looking to re-sort or re-partition your data by a different
>>> key or key combo after each output from reduce?
>>>
>>> On Fri, Oct 5, 2012 at 10:01 PM, Jim Twensky <[EMAIL PROTECTED]> wrote:
>>>> Hi,
>>>>
>>>> I have a complex Hadoop job that iterates over large graph data
>>>> multiple times until some convergence condition is met. I know that
>>>> the map output goes to the local disk of each particular mapper first,
>>>> and then fetched by the reducers before the reduce tasks start. I can
>>>> see that this is an overhead, and it theory we can ship the data
>>>> directly from mappers to reducers, without serializing on the local
>>>> disk first. I understand that this step is necessary for fault
>>>> tolerance and it is an essential building block of MapReduce.
>>>>
>>>> In my application, the map process consists of identity mappers which
>>>> read the input from HDFS and ship it to reducers. Essentially, what I
>>>> am doing is applying chains of reduce jobs until the algorithm
>>>> converges. My question is, can I bypass the serialization of the local
>>>> data and ship it from mappers to reducers immediately (as soon as I
>>>> call context.write() in my mapper class)? If not, are there any other
>>>> MR platforms that can do this? I've been searching around and couldn't
>>>> see anything similar to what I need. Hadoop On Line is a prototype and
>>>> has some similar functionality but it hasn't been updated for a while.
>>>>
>>>> Note: I know about ChainMapper and ChainReducer classes but I don't
>>>> want to chain multiple mappers in the same local node. I want to chain
>>>> multiple reduce functions globally so the data flow looks like: Map ->
>>>> Reduce -> Reduce -> Reduce, which means each reduce operation is
>>>> followed by a shuffle and sort essentially bypassing the map
>>>> operation.
>>>
>>>
>>>
>>> --
>>> Harsh J
>
>
>
> --
> Harsh J
The hidden map operation which is applied to the reduced partition at
one stage can generate keys that are outside of the range covered by
that particular reducer. I still need to have the many-to-many
communication from reduce step k to reduce step k+1. Otherwise, I
think the ChainReducer would do the job and apply multiple maps to
each isolated partition produced by the reducer.
Jim
On Fri, Oct 5, 2012 at 12:54 PM, Harsh J <[EMAIL PROTECTED]> wrote:
> Would it then be right to assume that the keys produced by the reduced
> partition at one stage would be isolated to its partition alone and
> not occur in any of the other partition outputs? I'm guessing not,
> based on the nature of your data?
>
> I'm trying to understand why shuffling is good to be avoided here, and
> if it can be in some ways, given the data. As I see it, you need
> re-sort based on the new key per partition, but not the shuffle? Or am
> I wrong?
>
> On Fri, Oct 5, 2012 at 11:13 PM, Jim Twensky <[EMAIL PROTECTED]> wrote:
>> Hi Harsh,
>>
>> Yes, there is actually a "hidden" map stage, that generates new
>> <key,value> pairs based on the last reduce output but I can create
>> those records during the reduce step instead and get rid of the
>> intermediate map computation completely. The idea is to apply the map
>> function to each output of the reduce inside the reduce class and emit
>> the result as the output of the reducer.
>>
>> Jim
>>
>> On Fri, Oct 5, 2012 at 12:18 PM, Harsh J <[EMAIL PROTECTED]> wrote:
>>> Hey Jim,
>>>
>>> Are you looking to re-sort or re-partition your data by a different
>>> key or key combo after each output from reduce?
>>>
>>> On Fri, Oct 5, 2012 at 10:01 PM, Jim Twensky <[EMAIL PROTECTED]> wrote:
>>>> Hi,
>>>>
>>>> I have a complex Hadoop job that iterates over large graph data
>>>> multiple times until some convergence condition is met. I know that
>>>> the map output goes to the local disk of each particular mapper first,
>>>> and then fetched by the reducers before the reduce tasks start. I can
>>>> see that this is an overhead, and it theory we can ship the data
>>>> directly from mappers to reducers, without serializing on the local
>>>> disk first. I understand that this step is necessary for fault
>>>> tolerance and it is an essential building block of MapReduce.
>>>>
>>>> In my application, the map process consists of identity mappers which
>>>> read the input from HDFS and ship it to reducers. Essentially, what I
>>>> am doing is applying chains of reduce jobs until the algorithm
>>>> converges. My question is, can I bypass the serialization of the local
>>>> data and ship it from mappers to reducers immediately (as soon as I
>>>> call context.write() in my mapper class)? If not, are there any other
>>>> MR platforms that can do this? I've been searching around and couldn't
>>>> see anything similar to what I need. Hadoop On Line is a prototype and
>>>> has some similar functionality but it hasn't been updated for a while.
>>>>
>>>> Note: I know about ChainMapper and ChainReducer classes but I don't
>>>> want to chain multiple mappers in the same local node. I want to chain
>>>> multiple reduce functions globally so the data flow looks like: Map ->
>>>> Reduce -> Reduce -> Reduce, which means each reduce operation is
>>>> followed by a shuffle and sort essentially bypassing the map
>>>> operation.
>>>
>>>
>>>
>>> --
>>> Harsh J
>
>
>
> --
> Harsh J
-
Re: Chaning Multiple Reducers: Reduce -> Reduce -> ReduceBertrand Dechoux 2012-10-08, 10:39
Have you looked at graph processing for Hadoop? Like Hama (
http://hama.apache.org/) or Giraph (http://incubator.apache.org/giraph/).
I can't say for sure it would help you but it seems to be in the same
problem domain.
With regard to the chaining reducer issue this is indeed a general
implementation decision of Hadoop 1.
>From a purely functional point of view, regardless of performance, I guess
it could be shown that a map/reduce/map can be done with a reduce only and
that a sequence of map can be done with a single map. Of course, with
Hadoop the picture is bit more complex due to the sort phase.
map -> sort -> reduce : operations in map/reduce can not generally be
transferred due to the sort 'blocking' them when they are related to the
sort key
reduce -> map : all operations can be performed in the reduce
So
map -> sort -> reduce -> map -> sort -> reduce -> map -> sort -> reduce
can generally be implemented as
map -> sort -> reduce -> sort -> reduce -> sort -> reduce
if you are willing to let the possibility of having different scaling
options for maps and reduces
And that's what you are asking. But with hadoop 1 the map phase is not an
option (even though you could use the identify but that's not a wise option
with regards to performance like you said). The picture might be changing
with Hadoop 2/YARN. I can't provide the details but it may be worth it to
look at it.
Regards
Bertrand
On Fri, Oct 5, 2012 at 8:02 PM, Jim Twensky <[EMAIL PROTECTED]> wrote:
> Hi Harsh,
>
> The hidden map operation which is applied to the reduced partition at
> one stage can generate keys that are outside of the range covered by
> that particular reducer. I still need to have the many-to-many
> communication from reduce step k to reduce step k+1. Otherwise, I
> think the ChainReducer would do the job and apply multiple maps to
> each isolated partition produced by the reducer.
>
> Jim
>
> On Fri, Oct 5, 2012 at 12:54 PM, Harsh J <[EMAIL PROTECTED]> wrote:
> > Would it then be right to assume that the keys produced by the reduced
> > partition at one stage would be isolated to its partition alone and
> > not occur in any of the other partition outputs? I'm guessing not,
> > based on the nature of your data?
> >
> > I'm trying to understand why shuffling is good to be avoided here, and
> > if it can be in some ways, given the data. As I see it, you need
> > re-sort based on the new key per partition, but not the shuffle? Or am
> > I wrong?
> >
> > On Fri, Oct 5, 2012 at 11:13 PM, Jim Twensky <[EMAIL PROTECTED]>
> wrote:
> >> Hi Harsh,
> >>
> >> Yes, there is actually a "hidden" map stage, that generates new
> >> <key,value> pairs based on the last reduce output but I can create
> >> those records during the reduce step instead and get rid of the
> >> intermediate map computation completely. The idea is to apply the map
> >> function to each output of the reduce inside the reduce class and emit
> >> the result as the output of the reducer.
> >>
> >> Jim
> >>
> >> On Fri, Oct 5, 2012 at 12:18 PM, Harsh J <[EMAIL PROTECTED]> wrote:
> >>> Hey Jim,
> >>>
> >>> Are you looking to re-sort or re-partition your data by a different
> >>> key or key combo after each output from reduce?
> >>>
> >>> On Fri, Oct 5, 2012 at 10:01 PM, Jim Twensky <[EMAIL PROTECTED]>
> wrote:
> >>>> Hi,
> >>>>
> >>>> I have a complex Hadoop job that iterates over large graph data
> >>>> multiple times until some convergence condition is met. I know that
> >>>> the map output goes to the local disk of each particular mapper first,
> >>>> and then fetched by the reducers before the reduce tasks start. I can
> >>>> see that this is an overhead, and it theory we can ship the data
> >>>> directly from mappers to reducers, without serializing on the local
> >>>> disk first. I understand that this step is necessary for fault
> >>>> tolerance and it is an essential building block of MapReduce.
> >>>>
> >>>> In my application, the map process consists of identity mappers which
Bertrand Dechoux
http://hama.apache.org/) or Giraph (http://incubator.apache.org/giraph/).
I can't say for sure it would help you but it seems to be in the same
problem domain.
With regard to the chaining reducer issue this is indeed a general
implementation decision of Hadoop 1.
>From a purely functional point of view, regardless of performance, I guess
it could be shown that a map/reduce/map can be done with a reduce only and
that a sequence of map can be done with a single map. Of course, with
Hadoop the picture is bit more complex due to the sort phase.
map -> sort -> reduce : operations in map/reduce can not generally be
transferred due to the sort 'blocking' them when they are related to the
sort key
reduce -> map : all operations can be performed in the reduce
So
map -> sort -> reduce -> map -> sort -> reduce -> map -> sort -> reduce
can generally be implemented as
map -> sort -> reduce -> sort -> reduce -> sort -> reduce
if you are willing to let the possibility of having different scaling
options for maps and reduces
And that's what you are asking. But with hadoop 1 the map phase is not an
option (even though you could use the identify but that's not a wise option
with regards to performance like you said). The picture might be changing
with Hadoop 2/YARN. I can't provide the details but it may be worth it to
look at it.
Regards
Bertrand
On Fri, Oct 5, 2012 at 8:02 PM, Jim Twensky <[EMAIL PROTECTED]> wrote:
> Hi Harsh,
>
> The hidden map operation which is applied to the reduced partition at
> one stage can generate keys that are outside of the range covered by
> that particular reducer. I still need to have the many-to-many
> communication from reduce step k to reduce step k+1. Otherwise, I
> think the ChainReducer would do the job and apply multiple maps to
> each isolated partition produced by the reducer.
>
> Jim
>
> On Fri, Oct 5, 2012 at 12:54 PM, Harsh J <[EMAIL PROTECTED]> wrote:
> > Would it then be right to assume that the keys produced by the reduced
> > partition at one stage would be isolated to its partition alone and
> > not occur in any of the other partition outputs? I'm guessing not,
> > based on the nature of your data?
> >
> > I'm trying to understand why shuffling is good to be avoided here, and
> > if it can be in some ways, given the data. As I see it, you need
> > re-sort based on the new key per partition, but not the shuffle? Or am
> > I wrong?
> >
> > On Fri, Oct 5, 2012 at 11:13 PM, Jim Twensky <[EMAIL PROTECTED]>
> wrote:
> >> Hi Harsh,
> >>
> >> Yes, there is actually a "hidden" map stage, that generates new
> >> <key,value> pairs based on the last reduce output but I can create
> >> those records during the reduce step instead and get rid of the
> >> intermediate map computation completely. The idea is to apply the map
> >> function to each output of the reduce inside the reduce class and emit
> >> the result as the output of the reducer.
> >>
> >> Jim
> >>
> >> On Fri, Oct 5, 2012 at 12:18 PM, Harsh J <[EMAIL PROTECTED]> wrote:
> >>> Hey Jim,
> >>>
> >>> Are you looking to re-sort or re-partition your data by a different
> >>> key or key combo after each output from reduce?
> >>>
> >>> On Fri, Oct 5, 2012 at 10:01 PM, Jim Twensky <[EMAIL PROTECTED]>
> wrote:
> >>>> Hi,
> >>>>
> >>>> I have a complex Hadoop job that iterates over large graph data
> >>>> multiple times until some convergence condition is met. I know that
> >>>> the map output goes to the local disk of each particular mapper first,
> >>>> and then fetched by the reducers before the reduce tasks start. I can
> >>>> see that this is an overhead, and it theory we can ship the data
> >>>> directly from mappers to reducers, without serializing on the local
> >>>> disk first. I understand that this step is necessary for fault
> >>>> tolerance and it is an essential building block of MapReduce.
> >>>>
> >>>> In my application, the map process consists of identity mappers which
Bertrand Dechoux
-
Re: Chaning Multiple Reducers: Reduce -> Reduce -> ReduceFabio Pitzolu 2012-10-08, 10:44
Isn't also of some help using Cascading (http://www.cascading.org/) ?
*Fabio Pitzolu*
Consultant - BI & Infrastructure
Mob. +39 3356033776
Telefono 02 87157239
Fax. 02 93664786
*Gruppo Consulenza Innovazione - http://www.gr-ci.com*
2012/10/8 Bertrand Dechoux <[EMAIL PROTECTED]>
> Have you looked at graph processing for Hadoop? Like Hama (
> http://hama.apache.org/) or Giraph (http://incubator.apache.org/giraph/).
> I can't say for sure it would help you but it seems to be in the same
> problem domain.
>
> With regard to the chaining reducer issue this is indeed a general
> implementation decision of Hadoop 1.
> From a purely functional point of view, regardless of performance, I guess
> it could be shown that a map/reduce/map can be done with a reduce only and
> that a sequence of map can be done with a single map. Of course, with
> Hadoop the picture is bit more complex due to the sort phase.
>
> map -> sort -> reduce : operations in map/reduce can not generally be
> transferred due to the sort 'blocking' them when they are related to the
> sort key
> reduce -> map : all operations can be performed in the reduce
> So
> map -> sort -> reduce -> map -> sort -> reduce -> map -> sort -> reduce
> can generally be implemented as
> map -> sort -> reduce -> sort -> reduce -> sort -> reduce
> if you are willing to let the possibility of having different scaling
> options for maps and reduces
>
> And that's what you are asking. But with hadoop 1 the map phase is not an
> option (even though you could use the identify but that's not a wise option
> with regards to performance like you said). The picture might be changing
> with Hadoop 2/YARN. I can't provide the details but it may be worth it to
> look at it.
>
> Regards
>
> Bertrand
>
>
> On Fri, Oct 5, 2012 at 8:02 PM, Jim Twensky <[EMAIL PROTECTED]> wrote:
>
>> Hi Harsh,
>>
>> The hidden map operation which is applied to the reduced partition at
>> one stage can generate keys that are outside of the range covered by
>> that particular reducer. I still need to have the many-to-many
>> communication from reduce step k to reduce step k+1. Otherwise, I
>> think the ChainReducer would do the job and apply multiple maps to
>> each isolated partition produced by the reducer.
>>
>> Jim
>>
>> On Fri, Oct 5, 2012 at 12:54 PM, Harsh J <[EMAIL PROTECTED]> wrote:
>> > Would it then be right to assume that the keys produced by the reduced
>> > partition at one stage would be isolated to its partition alone and
>> > not occur in any of the other partition outputs? I'm guessing not,
>> > based on the nature of your data?
>> >
>> > I'm trying to understand why shuffling is good to be avoided here, and
>> > if it can be in some ways, given the data. As I see it, you need
>> > re-sort based on the new key per partition, but not the shuffle? Or am
>> > I wrong?
>> >
>> > On Fri, Oct 5, 2012 at 11:13 PM, Jim Twensky <[EMAIL PROTECTED]>
>> wrote:
>> >> Hi Harsh,
>> >>
>> >> Yes, there is actually a "hidden" map stage, that generates new
>> >> <key,value> pairs based on the last reduce output but I can create
>> >> those records during the reduce step instead and get rid of the
>> >> intermediate map computation completely. The idea is to apply the map
>> >> function to each output of the reduce inside the reduce class and emit
>> >> the result as the output of the reducer.
>> >>
>> >> Jim
>> >>
>> >> On Fri, Oct 5, 2012 at 12:18 PM, Harsh J <[EMAIL PROTECTED]> wrote:
>> >>> Hey Jim,
>> >>>
>> >>> Are you looking to re-sort or re-partition your data by a different
>> >>> key or key combo after each output from reduce?
>> >>>
>> >>> On Fri, Oct 5, 2012 at 10:01 PM, Jim Twensky <[EMAIL PROTECTED]>
>> wrote:
>> >>>> Hi,
>> >>>>
>> >>>> I have a complex Hadoop job that iterates over large graph data
>> >>>> multiple times until some convergence condition is met. I know that
>> >>>> the map output goes to the local disk of each particular mapper
>> first,
>> >>>> and then fetched by the reducers before the reduce tasks start. I can
*Fabio Pitzolu*
Consultant - BI & Infrastructure
Mob. +39 3356033776
Telefono 02 87157239
Fax. 02 93664786
*Gruppo Consulenza Innovazione - http://www.gr-ci.com*
2012/10/8 Bertrand Dechoux <[EMAIL PROTECTED]>
> Have you looked at graph processing for Hadoop? Like Hama (
> http://hama.apache.org/) or Giraph (http://incubator.apache.org/giraph/).
> I can't say for sure it would help you but it seems to be in the same
> problem domain.
>
> With regard to the chaining reducer issue this is indeed a general
> implementation decision of Hadoop 1.
> From a purely functional point of view, regardless of performance, I guess
> it could be shown that a map/reduce/map can be done with a reduce only and
> that a sequence of map can be done with a single map. Of course, with
> Hadoop the picture is bit more complex due to the sort phase.
>
> map -> sort -> reduce : operations in map/reduce can not generally be
> transferred due to the sort 'blocking' them when they are related to the
> sort key
> reduce -> map : all operations can be performed in the reduce
> So
> map -> sort -> reduce -> map -> sort -> reduce -> map -> sort -> reduce
> can generally be implemented as
> map -> sort -> reduce -> sort -> reduce -> sort -> reduce
> if you are willing to let the possibility of having different scaling
> options for maps and reduces
>
> And that's what you are asking. But with hadoop 1 the map phase is not an
> option (even though you could use the identify but that's not a wise option
> with regards to performance like you said). The picture might be changing
> with Hadoop 2/YARN. I can't provide the details but it may be worth it to
> look at it.
>
> Regards
>
> Bertrand
>
>
> On Fri, Oct 5, 2012 at 8:02 PM, Jim Twensky <[EMAIL PROTECTED]> wrote:
>
>> Hi Harsh,
>>
>> The hidden map operation which is applied to the reduced partition at
>> one stage can generate keys that are outside of the range covered by
>> that particular reducer. I still need to have the many-to-many
>> communication from reduce step k to reduce step k+1. Otherwise, I
>> think the ChainReducer would do the job and apply multiple maps to
>> each isolated partition produced by the reducer.
>>
>> Jim
>>
>> On Fri, Oct 5, 2012 at 12:54 PM, Harsh J <[EMAIL PROTECTED]> wrote:
>> > Would it then be right to assume that the keys produced by the reduced
>> > partition at one stage would be isolated to its partition alone and
>> > not occur in any of the other partition outputs? I'm guessing not,
>> > based on the nature of your data?
>> >
>> > I'm trying to understand why shuffling is good to be avoided here, and
>> > if it can be in some ways, given the data. As I see it, you need
>> > re-sort based on the new key per partition, but not the shuffle? Or am
>> > I wrong?
>> >
>> > On Fri, Oct 5, 2012 at 11:13 PM, Jim Twensky <[EMAIL PROTECTED]>
>> wrote:
>> >> Hi Harsh,
>> >>
>> >> Yes, there is actually a "hidden" map stage, that generates new
>> >> <key,value> pairs based on the last reduce output but I can create
>> >> those records during the reduce step instead and get rid of the
>> >> intermediate map computation completely. The idea is to apply the map
>> >> function to each output of the reduce inside the reduce class and emit
>> >> the result as the output of the reducer.
>> >>
>> >> Jim
>> >>
>> >> On Fri, Oct 5, 2012 at 12:18 PM, Harsh J <[EMAIL PROTECTED]> wrote:
>> >>> Hey Jim,
>> >>>
>> >>> Are you looking to re-sort or re-partition your data by a different
>> >>> key or key combo after each output from reduce?
>> >>>
>> >>> On Fri, Oct 5, 2012 at 10:01 PM, Jim Twensky <[EMAIL PROTECTED]>
>> wrote:
>> >>>> Hi,
>> >>>>
>> >>>> I have a complex Hadoop job that iterates over large graph data
>> >>>> multiple times until some convergence condition is met. I know that
>> >>>> the map output goes to the local disk of each particular mapper
>> first,
>> >>>> and then fetched by the reducers before the reduce tasks start. I can
-
Re: Chaning Multiple Reducers: Reduce -> Reduce -> ReduceBertrand Dechoux 2012-10-08, 10:51
The question is not how to sequence all. Cascading could indeed help in
that case.
But how to skip the map phase and do the split/local sort directly at the
end of the reduce so that the next reduce need only to do a merge on the
sorted files obtained from the previous reduce. This is basically a
performance optimization (avoid unnecessary network/disk transfers).
Cascading is not equipped to do it, it will only compile the flow into a
sequence of map-reduce.
Regards
Bertrand
On Mon, Oct 8, 2012 at 12:44 PM, Fabio Pitzolu <[EMAIL PROTECTED]>wrote:
> Isn't also of some help using Cascading (http://www.cascading.org/) ?
>
> *Fabio Pitzolu*
> Consultant - BI & Infrastructure
>
> Mob. +39 3356033776
> Telefono 02 87157239
> Fax. 02 93664786
>
> *Gruppo Consulenza Innovazione - http://www.gr-ci.com*
>
>
>
>
> 2012/10/8 Bertrand Dechoux <[EMAIL PROTECTED]>
>
>> Have you looked at graph processing for Hadoop? Like Hama (
>> http://hama.apache.org/) or Giraph (http://incubator.apache.org/giraph/).
>> I can't say for sure it would help you but it seems to be in the same
>> problem domain.
>>
>> With regard to the chaining reducer issue this is indeed a general
>> implementation decision of Hadoop 1.
>> From a purely functional point of view, regardless of performance, I
>> guess it could be shown that a map/reduce/map can be done with a reduce
>> only and that a sequence of map can be done with a single map. Of course,
>> with Hadoop the picture is bit more complex due to the sort phase.
>>
>> map -> sort -> reduce : operations in map/reduce can not generally be
>> transferred due to the sort 'blocking' them when they are related to the
>> sort key
>> reduce -> map : all operations can be performed in the reduce
>> So
>> map -> sort -> reduce -> map -> sort -> reduce -> map -> sort -> reduce
>> can generally be implemented as
>> map -> sort -> reduce -> sort -> reduce -> sort -> reduce
>> if you are willing to let the possibility of having different scaling
>> options for maps and reduces
>>
>> And that's what you are asking. But with hadoop 1 the map phase is not an
>> option (even though you could use the identify but that's not a wise option
>> with regards to performance like you said). The picture might be changing
>> with Hadoop 2/YARN. I can't provide the details but it may be worth it to
>> look at it.
>>
>> Regards
>>
>> Bertrand
>>
>>
>> On Fri, Oct 5, 2012 at 8:02 PM, Jim Twensky <[EMAIL PROTECTED]>wrote:
>>
>>> Hi Harsh,
>>>
>>> The hidden map operation which is applied to the reduced partition at
>>> one stage can generate keys that are outside of the range covered by
>>> that particular reducer. I still need to have the many-to-many
>>> communication from reduce step k to reduce step k+1. Otherwise, I
>>> think the ChainReducer would do the job and apply multiple maps to
>>> each isolated partition produced by the reducer.
>>>
>>> Jim
>>>
>>> On Fri, Oct 5, 2012 at 12:54 PM, Harsh J <[EMAIL PROTECTED]> wrote:
>>> > Would it then be right to assume that the keys produced by the reduced
>>> > partition at one stage would be isolated to its partition alone and
>>> > not occur in any of the other partition outputs? I'm guessing not,
>>> > based on the nature of your data?
>>> >
>>> > I'm trying to understand why shuffling is good to be avoided here, and
>>> > if it can be in some ways, given the data. As I see it, you need
>>> > re-sort based on the new key per partition, but not the shuffle? Or am
>>> > I wrong?
>>> >
>>> > On Fri, Oct 5, 2012 at 11:13 PM, Jim Twensky <[EMAIL PROTECTED]>
>>> wrote:
>>> >> Hi Harsh,
>>> >>
>>> >> Yes, there is actually a "hidden" map stage, that generates new
>>> >> <key,value> pairs based on the last reduce output but I can create
>>> >> those records during the reduce step instead and get rid of the
>>> >> intermediate map computation completely. The idea is to apply the map
>>> >> function to each output of the reduce inside the reduce class and emit
>>> >> the result as the output of the reducer.
Bertrand Dechoux
that case.
But how to skip the map phase and do the split/local sort directly at the
end of the reduce so that the next reduce need only to do a merge on the
sorted files obtained from the previous reduce. This is basically a
performance optimization (avoid unnecessary network/disk transfers).
Cascading is not equipped to do it, it will only compile the flow into a
sequence of map-reduce.
Regards
Bertrand
On Mon, Oct 8, 2012 at 12:44 PM, Fabio Pitzolu <[EMAIL PROTECTED]>wrote:
> Isn't also of some help using Cascading (http://www.cascading.org/) ?
>
> *Fabio Pitzolu*
> Consultant - BI & Infrastructure
>
> Mob. +39 3356033776
> Telefono 02 87157239
> Fax. 02 93664786
>
> *Gruppo Consulenza Innovazione - http://www.gr-ci.com*
>
>
>
>
> 2012/10/8 Bertrand Dechoux <[EMAIL PROTECTED]>
>
>> Have you looked at graph processing for Hadoop? Like Hama (
>> http://hama.apache.org/) or Giraph (http://incubator.apache.org/giraph/).
>> I can't say for sure it would help you but it seems to be in the same
>> problem domain.
>>
>> With regard to the chaining reducer issue this is indeed a general
>> implementation decision of Hadoop 1.
>> From a purely functional point of view, regardless of performance, I
>> guess it could be shown that a map/reduce/map can be done with a reduce
>> only and that a sequence of map can be done with a single map. Of course,
>> with Hadoop the picture is bit more complex due to the sort phase.
>>
>> map -> sort -> reduce : operations in map/reduce can not generally be
>> transferred due to the sort 'blocking' them when they are related to the
>> sort key
>> reduce -> map : all operations can be performed in the reduce
>> So
>> map -> sort -> reduce -> map -> sort -> reduce -> map -> sort -> reduce
>> can generally be implemented as
>> map -> sort -> reduce -> sort -> reduce -> sort -> reduce
>> if you are willing to let the possibility of having different scaling
>> options for maps and reduces
>>
>> And that's what you are asking. But with hadoop 1 the map phase is not an
>> option (even though you could use the identify but that's not a wise option
>> with regards to performance like you said). The picture might be changing
>> with Hadoop 2/YARN. I can't provide the details but it may be worth it to
>> look at it.
>>
>> Regards
>>
>> Bertrand
>>
>>
>> On Fri, Oct 5, 2012 at 8:02 PM, Jim Twensky <[EMAIL PROTECTED]>wrote:
>>
>>> Hi Harsh,
>>>
>>> The hidden map operation which is applied to the reduced partition at
>>> one stage can generate keys that are outside of the range covered by
>>> that particular reducer. I still need to have the many-to-many
>>> communication from reduce step k to reduce step k+1. Otherwise, I
>>> think the ChainReducer would do the job and apply multiple maps to
>>> each isolated partition produced by the reducer.
>>>
>>> Jim
>>>
>>> On Fri, Oct 5, 2012 at 12:54 PM, Harsh J <[EMAIL PROTECTED]> wrote:
>>> > Would it then be right to assume that the keys produced by the reduced
>>> > partition at one stage would be isolated to its partition alone and
>>> > not occur in any of the other partition outputs? I'm guessing not,
>>> > based on the nature of your data?
>>> >
>>> > I'm trying to understand why shuffling is good to be avoided here, and
>>> > if it can be in some ways, given the data. As I see it, you need
>>> > re-sort based on the new key per partition, but not the shuffle? Or am
>>> > I wrong?
>>> >
>>> > On Fri, Oct 5, 2012 at 11:13 PM, Jim Twensky <[EMAIL PROTECTED]>
>>> wrote:
>>> >> Hi Harsh,
>>> >>
>>> >> Yes, there is actually a "hidden" map stage, that generates new
>>> >> <key,value> pairs based on the last reduce output but I can create
>>> >> those records during the reduce step instead and get rid of the
>>> >> intermediate map computation completely. The idea is to apply the map
>>> >> function to each output of the reduce inside the reduce class and emit
>>> >> the result as the output of the reducer.
Bertrand Dechoux
-
Re: Chaning Multiple Reducers: Reduce -> Reduce -> ReduceJim Twensky 2012-10-08, 19:09
Thank you for the comments. Some similar frameworks I looked at
include Haloop, Twister, Hama, Giraph and Cascading. I am also doing
large scale graph processing so I assumed one of them could serve the
purpose. Here is a summary of what I found out about them that is
relevant:
1) Haloop and Twister: They cache static data among a chain of
MapReduce jobs. The main contribution is to reduce the intermediate
data shipped from mappers to reducers. Still, the output of each
reduce goes to the file system.
2) Cascading: A higher level API to create MapReduce workflows.
Anything you can do with Cascading can be done practically by more
programing effort and using Hadoop only. Bypassing map and running a
chain of sort->reduce->sort->reduce jobs is not possible. Please
correct me if I'm wrong.
3) Giraph: Built on the BSP model and is very similar to Pregel. I
couldn't find a detailed overview of their architecture but my
understanding is that your data needs to fit in distributed memory,
which is also true for Pregel.
4) Hama: Also follows the BSP model. I don't know how the intermediate
data is serialized and passed to the next set of nodes and whether it
is possible to do a performance optimization similar to what I am
asking for. If anyone who used Hama can point a few articles about how
the framework actually works and handles the messages passed between
vertices, I'd really appreciate that.
Conclusion: None of the above tools can bypass the map step or do a
similar performance optimization. Of course Giraph and Hama are built
on a different model - not really MapReduce - so it is not very
accurate to say that they don't have the required functionality.
If I'm missing anything and.or if there are folks who used Giraph or
Hama and think that they might serve the purpose, I'd be glad to hear
more.
Jim
On Mon, Oct 8, 2012 at 6:52 AM, Michael Segel <[EMAIL PROTECTED]> wrote:
> I don't believe that Hama would suffice.
>
> In terms of M/R where you want to chain reducers...
> Can you chain combiners? (I don't think so, but you never know)
>
> If not, you end up with a series of M/R jobs and the Mappers are just identity mappers.
>
> Or you could use HBase, with a small caveat... you have to be careful not to use speculative execution and that if a task fails, that the results of the task won't be affected if they are run a second time. Meaning that they will just overwrite the data in a column with a second cell and that you don't care about the number of versions.
>
> Note: HBase doesn't have transactions, so you would have to think about how to tag cells so that if a task dies, upon restart, you can remove the affected cells. Along with some post job synchronization...
>
> Again HBase may work, but there may also be additional problems that could impact your results. It will have to be evaluated on a case by case basis.
>
>
> JMHO
>
> -Mike
>
> On Oct 8, 2012, at 6:35 AM, Edward J. Yoon <[EMAIL PROTECTED]> wrote:
>
>>> call context.write() in my mapper class)? If not, are there any other
>>> MR platforms that can do this? I've been searching around and couldn't
>>
>> You can use Hama BSP[1] instead of Map/Reduce.
>>
>> No stable release yet but I confirmed that large graph with billions
>> of nodes and edges can be crunched in few minutes[2].
>>
>> 1. http://hama.apache.org
>> 2. http://wiki.apache.org/hama/Benchmarks
>>
>> On Sat, Oct 6, 2012 at 1:31 AM, Jim Twensky <[EMAIL PROTECTED]> wrote:
>>> Hi,
>>>
>>> I have a complex Hadoop job that iterates over large graph data
>>> multiple times until some convergence condition is met. I know that
>>> the map output goes to the local disk of each particular mapper first,
>>> and then fetched by the reducers before the reduce tasks start. I can
>>> see that this is an overhead, and it theory we can ship the data
>>> directly from mappers to reducers, without serializing on the local
>>> disk first. I understand that this step is necessary for fault
>>> tolerance and it is an essential building block of MapReduce.
include Haloop, Twister, Hama, Giraph and Cascading. I am also doing
large scale graph processing so I assumed one of them could serve the
purpose. Here is a summary of what I found out about them that is
relevant:
1) Haloop and Twister: They cache static data among a chain of
MapReduce jobs. The main contribution is to reduce the intermediate
data shipped from mappers to reducers. Still, the output of each
reduce goes to the file system.
2) Cascading: A higher level API to create MapReduce workflows.
Anything you can do with Cascading can be done practically by more
programing effort and using Hadoop only. Bypassing map and running a
chain of sort->reduce->sort->reduce jobs is not possible. Please
correct me if I'm wrong.
3) Giraph: Built on the BSP model and is very similar to Pregel. I
couldn't find a detailed overview of their architecture but my
understanding is that your data needs to fit in distributed memory,
which is also true for Pregel.
4) Hama: Also follows the BSP model. I don't know how the intermediate
data is serialized and passed to the next set of nodes and whether it
is possible to do a performance optimization similar to what I am
asking for. If anyone who used Hama can point a few articles about how
the framework actually works and handles the messages passed between
vertices, I'd really appreciate that.
Conclusion: None of the above tools can bypass the map step or do a
similar performance optimization. Of course Giraph and Hama are built
on a different model - not really MapReduce - so it is not very
accurate to say that they don't have the required functionality.
If I'm missing anything and.or if there are folks who used Giraph or
Hama and think that they might serve the purpose, I'd be glad to hear
more.
Jim
On Mon, Oct 8, 2012 at 6:52 AM, Michael Segel <[EMAIL PROTECTED]> wrote:
> I don't believe that Hama would suffice.
>
> In terms of M/R where you want to chain reducers...
> Can you chain combiners? (I don't think so, but you never know)
>
> If not, you end up with a series of M/R jobs and the Mappers are just identity mappers.
>
> Or you could use HBase, with a small caveat... you have to be careful not to use speculative execution and that if a task fails, that the results of the task won't be affected if they are run a second time. Meaning that they will just overwrite the data in a column with a second cell and that you don't care about the number of versions.
>
> Note: HBase doesn't have transactions, so you would have to think about how to tag cells so that if a task dies, upon restart, you can remove the affected cells. Along with some post job synchronization...
>
> Again HBase may work, but there may also be additional problems that could impact your results. It will have to be evaluated on a case by case basis.
>
>
> JMHO
>
> -Mike
>
> On Oct 8, 2012, at 6:35 AM, Edward J. Yoon <[EMAIL PROTECTED]> wrote:
>
>>> call context.write() in my mapper class)? If not, are there any other
>>> MR platforms that can do this? I've been searching around and couldn't
>>
>> You can use Hama BSP[1] instead of Map/Reduce.
>>
>> No stable release yet but I confirmed that large graph with billions
>> of nodes and edges can be crunched in few minutes[2].
>>
>> 1. http://hama.apache.org
>> 2. http://wiki.apache.org/hama/Benchmarks
>>
>> On Sat, Oct 6, 2012 at 1:31 AM, Jim Twensky <[EMAIL PROTECTED]> wrote:
>>> Hi,
>>>
>>> I have a complex Hadoop job that iterates over large graph data
>>> multiple times until some convergence condition is met. I know that
>>> the map output goes to the local disk of each particular mapper first,
>>> and then fetched by the reducers before the reduce tasks start. I can
>>> see that this is an overhead, and it theory we can ship the data
>>> directly from mappers to reducers, without serializing on the local
>>> disk first. I understand that this step is necessary for fault
>>> tolerance and it is an essential building block of MapReduce.
-
Re: Chaning Multiple Reducers: Reduce -> Reduce -> ReduceMichael Segel 2012-10-08, 19:19
Well I was thinking ...
Map -> Combiner -> Reducer -> Identity Mapper -> combiner -> reducer -> Identity Mapper -> combiner -> reducer...
May make things easier.
HTH
0Mike
On Oct 8, 2012, at 2:09 PM, Jim Twensky <[EMAIL PROTECTED]> wrote:
> Thank you for the comments. Some similar frameworks I looked at
> include Haloop, Twister, Hama, Giraph and Cascading. I am also doing
> large scale graph processing so I assumed one of them could serve the
> purpose. Here is a summary of what I found out about them that is
> relevant:
>
> 1) Haloop and Twister: They cache static data among a chain of
> MapReduce jobs. The main contribution is to reduce the intermediate
> data shipped from mappers to reducers. Still, the output of each
> reduce goes to the file system.
>
> 2) Cascading: A higher level API to create MapReduce workflows.
> Anything you can do with Cascading can be done practically by more
> programing effort and using Hadoop only. Bypassing map and running a
> chain of sort->reduce->sort->reduce jobs is not possible. Please
> correct me if I'm wrong.
>
> 3) Giraph: Built on the BSP model and is very similar to Pregel. I
> couldn't find a detailed overview of their architecture but my
> understanding is that your data needs to fit in distributed memory,
> which is also true for Pregel.
>
> 4) Hama: Also follows the BSP model. I don't know how the intermediate
> data is serialized and passed to the next set of nodes and whether it
> is possible to do a performance optimization similar to what I am
> asking for. If anyone who used Hama can point a few articles about how
> the framework actually works and handles the messages passed between
> vertices, I'd really appreciate that.
>
> Conclusion: None of the above tools can bypass the map step or do a
> similar performance optimization. Of course Giraph and Hama are built
> on a different model - not really MapReduce - so it is not very
> accurate to say that they don't have the required functionality.
>
> If I'm missing anything and.or if there are folks who used Giraph or
> Hama and think that they might serve the purpose, I'd be glad to hear
> more.
>
> Jim
>
> On Mon, Oct 8, 2012 at 6:52 AM, Michael Segel <[EMAIL PROTECTED]> wrote:
>> I don't believe that Hama would suffice.
>>
>> In terms of M/R where you want to chain reducers...
>> Can you chain combiners? (I don't think so, but you never know)
>>
>> If not, you end up with a series of M/R jobs and the Mappers are just identity mappers.
>>
>> Or you could use HBase, with a small caveat... you have to be careful not to use speculative execution and that if a task fails, that the results of the task won't be affected if they are run a second time. Meaning that they will just overwrite the data in a column with a second cell and that you don't care about the number of versions.
>>
>> Note: HBase doesn't have transactions, so you would have to think about how to tag cells so that if a task dies, upon restart, you can remove the affected cells. Along with some post job synchronization...
>>
>> Again HBase may work, but there may also be additional problems that could impact your results. It will have to be evaluated on a case by case basis.
>>
>>
>> JMHO
>>
>> -Mike
>>
>> On Oct 8, 2012, at 6:35 AM, Edward J. Yoon <[EMAIL PROTECTED]> wrote:
>>
>>>> call context.write() in my mapper class)? If not, are there any other
>>>> MR platforms that can do this? I've been searching around and couldn't
>>>
>>> You can use Hama BSP[1] instead of Map/Reduce.
>>>
>>> No stable release yet but I confirmed that large graph with billions
>>> of nodes and edges can be crunched in few minutes[2].
>>>
>>> 1. http://hama.apache.org
>>> 2. http://wiki.apache.org/hama/Benchmarks
>>>
>>> On Sat, Oct 6, 2012 at 1:31 AM, Jim Twensky <[EMAIL PROTECTED]> wrote:
>>>> Hi,
>>>>
>>>> I have a complex Hadoop job that iterates over large graph data
>>>> multiple times until some convergence condition is met. I know that
Map -> Combiner -> Reducer -> Identity Mapper -> combiner -> reducer -> Identity Mapper -> combiner -> reducer...
May make things easier.
HTH
0Mike
On Oct 8, 2012, at 2:09 PM, Jim Twensky <[EMAIL PROTECTED]> wrote:
> Thank you for the comments. Some similar frameworks I looked at
> include Haloop, Twister, Hama, Giraph and Cascading. I am also doing
> large scale graph processing so I assumed one of them could serve the
> purpose. Here is a summary of what I found out about them that is
> relevant:
>
> 1) Haloop and Twister: They cache static data among a chain of
> MapReduce jobs. The main contribution is to reduce the intermediate
> data shipped from mappers to reducers. Still, the output of each
> reduce goes to the file system.
>
> 2) Cascading: A higher level API to create MapReduce workflows.
> Anything you can do with Cascading can be done practically by more
> programing effort and using Hadoop only. Bypassing map and running a
> chain of sort->reduce->sort->reduce jobs is not possible. Please
> correct me if I'm wrong.
>
> 3) Giraph: Built on the BSP model and is very similar to Pregel. I
> couldn't find a detailed overview of their architecture but my
> understanding is that your data needs to fit in distributed memory,
> which is also true for Pregel.
>
> 4) Hama: Also follows the BSP model. I don't know how the intermediate
> data is serialized and passed to the next set of nodes and whether it
> is possible to do a performance optimization similar to what I am
> asking for. If anyone who used Hama can point a few articles about how
> the framework actually works and handles the messages passed between
> vertices, I'd really appreciate that.
>
> Conclusion: None of the above tools can bypass the map step or do a
> similar performance optimization. Of course Giraph and Hama are built
> on a different model - not really MapReduce - so it is not very
> accurate to say that they don't have the required functionality.
>
> If I'm missing anything and.or if there are folks who used Giraph or
> Hama and think that they might serve the purpose, I'd be glad to hear
> more.
>
> Jim
>
> On Mon, Oct 8, 2012 at 6:52 AM, Michael Segel <[EMAIL PROTECTED]> wrote:
>> I don't believe that Hama would suffice.
>>
>> In terms of M/R where you want to chain reducers...
>> Can you chain combiners? (I don't think so, but you never know)
>>
>> If not, you end up with a series of M/R jobs and the Mappers are just identity mappers.
>>
>> Or you could use HBase, with a small caveat... you have to be careful not to use speculative execution and that if a task fails, that the results of the task won't be affected if they are run a second time. Meaning that they will just overwrite the data in a column with a second cell and that you don't care about the number of versions.
>>
>> Note: HBase doesn't have transactions, so you would have to think about how to tag cells so that if a task dies, upon restart, you can remove the affected cells. Along with some post job synchronization...
>>
>> Again HBase may work, but there may also be additional problems that could impact your results. It will have to be evaluated on a case by case basis.
>>
>>
>> JMHO
>>
>> -Mike
>>
>> On Oct 8, 2012, at 6:35 AM, Edward J. Yoon <[EMAIL PROTECTED]> wrote:
>>
>>>> call context.write() in my mapper class)? If not, are there any other
>>>> MR platforms that can do this? I've been searching around and couldn't
>>>
>>> You can use Hama BSP[1] instead of Map/Reduce.
>>>
>>> No stable release yet but I confirmed that large graph with billions
>>> of nodes and edges can be crunched in few minutes[2].
>>>
>>> 1. http://hama.apache.org
>>> 2. http://wiki.apache.org/hama/Benchmarks
>>>
>>> On Sat, Oct 6, 2012 at 1:31 AM, Jim Twensky <[EMAIL PROTECTED]> wrote:
>>>> Hi,
>>>>
>>>> I have a complex Hadoop job that iterates over large graph data
>>>> multiple times until some convergence condition is met. I know that
