You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@crunch.apache.org by Dave Beech <da...@paraliatech.com> on 2013/01/15 12:41:42 UTC
Question about mapreduce job planner
Hi all,
I've written a Crunch pipeline and have a question about the resulting
mapreduce jobs. Please see the steps below:
1. ) Load text data A and convert to avro -> A'
2. ) Load text data B and convert to avro -> B'
3. ) Union A' and B' -> C
4. ) Filter C -> D
5. ) Write D to HDFS
6a. ) Use DoFn to extract strings from D -> E
6b. ) Aggregate E ( count strings ) -> F
6c. ) Convert F to HBase puts -> G
6d. ) Write G to HBase
Running my code generates two mapreduce jobs which run in parallel:
job A) runs steps 1, 2, 3, 4, 5
job B) runs steps 1, 2, 3, 4, 6abcd
Without knowing much about the planning algorithm, what I expected to see
was more like:
job A) runs steps 1, 2, 3, 4, 5
job B) runs after A, reads back the data written in step 5 and does steps
6abcd.
So the jobs would run sequentially, not in parallel, but in doing so avoid
reading the full raw input data and performing the conversion/filtering
logic twice.
Is there a way I should order my pipeline calls or can I give hints to the
mapreduce compiler to do the jobs in this way? Does the scale factor have
any influence on this?
Thanks,
Dave
Re: Question about mapreduce job planner
Posted by Dave Beech <da...@paraliatech.com>.
Thanks Josh - that works. At least I was only 2 characters away from the
right answer! ;)
On 16 January 2013 15:40, Josh Wills <jw...@cloudera.com> wrote:
> Hey Dave,
>
> I forgot to tell you something important: your intermediate job should use
> At.avroFile(...) instead of To.avroFile(...) since you're planning on
> consuming additional data from it. If you do that, I believe it will work
> as expected (two sequential jobs with the second one picking up where the
> first one left off). In any case, we should make that transparent to users,
> so I'm writing a small patch to do the underlying Target -> SourceTarget
> conversion automatically when we can.
>
> Josh
>
>
> On Wed, Jan 16, 2013 at 2:34 AM, Dave Beech <da...@paraliatech.com> wrote:
>
>> Hi Josh. A follow up just to check I've got this straight.
>>
>> I've amended my pipeline and added a "pipeline.run()" call after the
>> write to HDFS. Now I do get two mapreduce jobs, but instead of the second
>> carrying on where the first left off, it actually re-does all the steps
>> needed to generate the PCollection that was written. I get the same jobs A
>> and B I described in my original email, but running sequentially rather
>> than in parallel. Is that what you'd expect?
>>
>> So I guess what I have to do following the write is re-read from the
>> output path using pipeline.read(From.avroFile(...)).
>>
>> It'd be good if the pipeline could hold onto information about
>> PCollections even after they're written, so that they can be used by
>> follow-on steps. I'll file a JIRA to this effect so we can discuss it
>> there.
>>
>> Thanks,
>> Dave
>>
>>
>> On 15 January 2013 21:00, Dave Beech <da...@paraliatech.com> wrote:
>>
>>> Thanks Josh - that's great. I'll file a JIRA about the side-outputs
>>> feature, but the pipeline.run() call will serve my purpose for now.
>>>
>>> Cheers,
>>> Dave
>>>
>>> On 15 January 2013 18:03, Josh Wills <jw...@cloudera.com> wrote:
>>>
>>>> Hey Dave,
>>>>
>>>> The way to force a sequential run would be to call pipeline.run() after
>>>> you write D to HDFS and before you declare the operations in step 6. What
>>>> we would really want here is a single MapReduce job that wrote side outputs
>>>> on the map side to create the dataset in step D, but we don't have support
>>>> for side-outputs in maps yet. Worth filing a JIRA, I think.
>>>>
>>>> Thanks!
>>>> Josh
>>>>
>>>
>>>
>>>
>>
>
>
> --
> Director of Data Science
> Cloudera <http://www.cloudera.com>
> Twitter: @josh_wills <http://twitter.com/josh_wills>
>
Re: Question about mapreduce job planner
Posted by Josh Wills <jw...@cloudera.com>.
Hey Dave,
I forgot to tell you something important: your intermediate job should use
At.avroFile(...) instead of To.avroFile(...) since you're planning on
consuming additional data from it. If you do that, I believe it will work
as expected (two sequential jobs with the second one picking up where the
first one left off). In any case, we should make that transparent to users,
so I'm writing a small patch to do the underlying Target -> SourceTarget
conversion automatically when we can.
Josh
On Wed, Jan 16, 2013 at 2:34 AM, Dave Beech <da...@paraliatech.com> wrote:
> Hi Josh. A follow up just to check I've got this straight.
>
> I've amended my pipeline and added a "pipeline.run()" call after the write
> to HDFS. Now I do get two mapreduce jobs, but instead of the second
> carrying on where the first left off, it actually re-does all the steps
> needed to generate the PCollection that was written. I get the same jobs A
> and B I described in my original email, but running sequentially rather
> than in parallel. Is that what you'd expect?
>
> So I guess what I have to do following the write is re-read from the
> output path using pipeline.read(From.avroFile(...)).
>
> It'd be good if the pipeline could hold onto information about
> PCollections even after they're written, so that they can be used by
> follow-on steps. I'll file a JIRA to this effect so we can discuss it
> there.
>
> Thanks,
> Dave
>
>
> On 15 January 2013 21:00, Dave Beech <da...@paraliatech.com> wrote:
>
>> Thanks Josh - that's great. I'll file a JIRA about the side-outputs
>> feature, but the pipeline.run() call will serve my purpose for now.
>>
>> Cheers,
>> Dave
>>
>> On 15 January 2013 18:03, Josh Wills <jw...@cloudera.com> wrote:
>>
>>> Hey Dave,
>>>
>>> The way to force a sequential run would be to call pipeline.run() after
>>> you write D to HDFS and before you declare the operations in step 6. What
>>> we would really want here is a single MapReduce job that wrote side outputs
>>> on the map side to create the dataset in step D, but we don't have support
>>> for side-outputs in maps yet. Worth filing a JIRA, I think.
>>>
>>> Thanks!
>>> Josh
>>>
>>
>>
>>
>
--
Director of Data Science
Cloudera <http://www.cloudera.com>
Twitter: @josh_wills <http://twitter.com/josh_wills>
Re: Question about mapreduce job planner
Posted by Dave Beech <da...@paraliatech.com>.
Hi Josh. A follow up just to check I've got this straight.
I've amended my pipeline and added a "pipeline.run()" call after the write
to HDFS. Now I do get two mapreduce jobs, but instead of the second
carrying on where the first left off, it actually re-does all the steps
needed to generate the PCollection that was written. I get the same jobs A
and B I described in my original email, but running sequentially rather
than in parallel. Is that what you'd expect?
So I guess what I have to do following the write is re-read from the output
path using pipeline.read(From.avroFile(...)).
It'd be good if the pipeline could hold onto information about PCollections
even after they're written, so that they can be used by follow-on steps.
I'll file a JIRA to this effect so we can discuss it there.
Thanks,
Dave
On 15 January 2013 21:00, Dave Beech <da...@paraliatech.com> wrote:
> Thanks Josh - that's great. I'll file a JIRA about the side-outputs
> feature, but the pipeline.run() call will serve my purpose for now.
>
> Cheers,
> Dave
>
> On 15 January 2013 18:03, Josh Wills <jw...@cloudera.com> wrote:
>
>> Hey Dave,
>>
>> The way to force a sequential run would be to call pipeline.run() after
>> you write D to HDFS and before you declare the operations in step 6. What
>> we would really want here is a single MapReduce job that wrote side outputs
>> on the map side to create the dataset in step D, but we don't have support
>> for side-outputs in maps yet. Worth filing a JIRA, I think.
>>
>> Thanks!
>> Josh
>>
>
>
>
Re: Question about mapreduce job planner
Posted by Dave Beech <da...@paraliatech.com>.
Thanks Josh - that's great. I'll file a JIRA about the side-outputs
feature, but the pipeline.run() call will serve my purpose for now.
Cheers,
Dave
On 15 January 2013 18:03, Josh Wills <jw...@cloudera.com> wrote:
> Hey Dave,
>
> The way to force a sequential run would be to call pipeline.run() after
> you write D to HDFS and before you declare the operations in step 6. What
> we would really want here is a single MapReduce job that wrote side outputs
> on the map side to create the dataset in step D, but we don't have support
> for side-outputs in maps yet. Worth filing a JIRA, I think.
>
> Thanks!
> Josh
>
Re: Question about mapreduce job planner
Posted by Josh Wills <jw...@cloudera.com>.
Hey Dave,
The way to force a sequential run would be to call pipeline.run() after you
write D to HDFS and before you declare the operations in step 6. What we
would really want here is a single MapReduce job that wrote side outputs on
the map side to create the dataset in step D, but we don't have support for
side-outputs in maps yet. Worth filing a JIRA, I think.
Thanks!
Josh
On Tue, Jan 15, 2013 at 3:41 AM, Dave Beech <da...@paraliatech.com> wrote:
> Hi all,
>
> I've written a Crunch pipeline and have a question about the resulting
> mapreduce jobs. Please see the steps below:
>
> 1. ) Load text data A and convert to avro -> A'
> 2. ) Load text data B and convert to avro -> B'
> 3. ) Union A' and B' -> C
> 4. ) Filter C -> D
>
> 5. ) Write D to HDFS
>
> 6a. ) Use DoFn to extract strings from D -> E
> 6b. ) Aggregate E ( count strings ) -> F
> 6c. ) Convert F to HBase puts -> G
> 6d. ) Write G to HBase
>
> Running my code generates two mapreduce jobs which run in parallel:
> job A) runs steps 1, 2, 3, 4, 5
> job B) runs steps 1, 2, 3, 4, 6abcd
>
> Without knowing much about the planning algorithm, what I expected to see
> was more like:
> job A) runs steps 1, 2, 3, 4, 5
> job B) runs after A, reads back the data written in step 5 and does steps
> 6abcd.
>
> So the jobs would run sequentially, not in parallel, but in doing so avoid
> reading the full raw input data and performing the conversion/filtering
> logic twice.
>
> Is there a way I should order my pipeline calls or can I give hints to the
> mapreduce compiler to do the jobs in this way? Does the scale factor have
> any influence on this?
>
> Thanks,
> Dave
>
--
Director of Data Science
Cloudera <http://www.cloudera.com>
Twitter: @josh_wills <http://twitter.com/josh_wills>