You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@crunch.apache.org by Rahul <rs...@xebia.com> on 2012/06/27 13:41:00 UTC
Exception due to same iterator returned back by PGroupedTableType
Hi,
I am trying to create multiple iterators in a DoFn process method.
/ public void process(Pair<Integer, Iterable<TupleN>> input,
Emitter<Pair<String, Integer>> emitter) {}
/Eve/ry /time I ask a iterator it gives back the same one and thus I
could not not traverse the list again and again as I am hitting the
following stack trace .
/
1575 [Thread-23] WARN org.apache.hadoop.mapred.LocalJobRunner -
job_local_0001
java.util.NoSuchElementException: iterate past last value
at
org.apache.hadoop.mapreduce.ReduceContext$ValueIterator.next(ReduceContext.java:159)
at
com.cloudera.crunch.types.PGroupedTableType$PTypeIterable$1.next(PGroupedTableType.java:56)
at
com.mylearning.crunch.TuplesTest$ScoreCalculator.process(TuplesTest.java:52)
at
com.mylearning.crunch.TuplesTest$ScoreCalculator.process(TuplesTest.java:1)
at com.cloudera.crunch.impl.mr.run.RTNode.process(RTNode.java:85)
at
com.cloudera.crunch.impl.mr.emit.IntermediateEmitter.emit(IntermediateEmitter.java:39)
.............../
What I see that /PGroupedTableType /returns back the same iterator
available to it via ReducerContext and this is the same reason that I
get to see the exception.
Shouldn't this give a back a new iterator everytime I ask it ?
If not then what should be the way of doing the same ? (I can iterate
first adding elements to a collection but then I donno if that should be
the way of going forward)
I have attached my test case for your reference.
regards,
Rahul
Re: Exception due to same iterator returned back by PGroupedTableType
Posted by Christian Tzolov <ch...@gmail.com>.
FYI, The "reusing the same iterator" semantics seems to be a common
pitfall. So I have extended the DoFn#process() documentation to explain the
impact imposed by the Hadoop Reducer implementation.
On Thu, Jun 28, 2012 at 1:25 PM, Josh Wills <jw...@cloudera.com> wrote:
> Hey Rahul,
>
> Re: groupByKey vs. collectValues: collectValues calls groupByKey in
> the course of its operations; it's basically just a convenience method
> for the kind of problem you're trying to solve (i.e., I need to
> iterate over the values returned by groupByKey multiple times, so
> please put them into a collection.) There are lots of other cases
> where you do not need to iterate over the values multiple times, and
> so Crunch (much like MapReduce) doesn't bother to keep everything
> around unless you explicitly ask it to do so.
>
> On Thu, Jun 28, 2012 at 4:13 AM, Rahul <rs...@xebia.com> wrote:
> > Hi Gabriel,
> >
> > I have found a way by which Crunch supports the uses case of having
> repeated
> > iterators but I am not completely sure of the in-outs of the same.
> > Basically rather than doing a groupBy on Ptable to get back a
> > PGroupedTable, I used the collectValues API to get back a
> > Ptable<key,Collection<values>>.
> >
> > PTable<Integer, Collection<TupleN>> collectValues =
> > classifiedData.collectValues();
> > PTable<String, Integer> scores = collectValues.parallelDo("compute
> > pairs",
> > new PTableScoreCalculator(),
> Writables.tableOf(Writables.strings(),
> > Writables.ints()));
> >
> >
> > Now when I do ParalledDo on the new collection I get back a Pair, having
> > keyType and ArrayList<valueType>, over which I can do things as I wish.
> >
> > class PTableScoreCalculator extends DoFn<Pair<Integer,
> Collection<TupleN>>,
> > Pair<String, Integer>> {
> > public void process(Pair<Integer, Collection<TupleN>> input,
> >
> > Emitter<Pair<String, Integer>> emitter) {
> > Iterator<TupleN> primary = input.second().iterator();
> > .....................
> > }
> >
> > This way I could iterate over again and again, any comments on the same.
> I
> > am attaching my test case for reference.
> >
> > BTW why are there two methods that can do the same things the groupBykey
> > method and the collectValues method ? I see an Aggregation gets invoked
> for
> > the collection API and in the other case a lazy collection gets created.
> Any
> > idea on the different applications of the two.
> >
> > regards,
> > Rahul
> >
> >
> >
> > On 28-06-2012 14:17, Gabriel Reid wrote:
> >
> > On Thu, Jun 28, 2012 at 9:29 AM, Rahul <rs...@xebia.com> wrote:
> >
> > Yes indeed this is a small PoC to get familiar with Crunch in relation
> to my
> > problem. Basically I have the following algo at play:
> > 1. Read data rows
> > 2. Create custom keys for each of them, built using various attributes of
> > data (this time it is just a simple hash code, but I would like to emit
> > multiple key-value pairs)
> > 3. Group similar data based on created Keys
> > 4. Iterate over individual items in the group and do extensive comparison
> > between all of them
> >
> > I just built an outline in the test case to see what/how can be done, can
> > you advise something better ?
> >
> > Thanks for the outline. In this case, your approach (with putting the
> > contents of the
> > incoming Iterable into a collection) should work fine, as long as
> > number of elements
> > per group is relatively small (i.e. easily able to fit in the memory
> > available to each reducer in your Hadoop cluster).
> >
> > - Gabriel
> >
> >
>
>
>
> --
> Director of Data Science
> Cloudera
> Twitter: @josh_wills
>
Re: Exception due to same iterator returned back by PGroupedTableType
Posted by Josh Wills <jw...@cloudera.com>.
Hey Rahul,
Re: groupByKey vs. collectValues: collectValues calls groupByKey in
the course of its operations; it's basically just a convenience method
for the kind of problem you're trying to solve (i.e., I need to
iterate over the values returned by groupByKey multiple times, so
please put them into a collection.) There are lots of other cases
where you do not need to iterate over the values multiple times, and
so Crunch (much like MapReduce) doesn't bother to keep everything
around unless you explicitly ask it to do so.
On Thu, Jun 28, 2012 at 4:13 AM, Rahul <rs...@xebia.com> wrote:
> Hi Gabriel,
>
> I have found a way by which Crunch supports the uses case of having repeated
> iterators but I am not completely sure of the in-outs of the same.
> Basically rather than doing a groupBy on Ptable to get back a
> PGroupedTable, I used the collectValues API to get back a
> Ptable<key,Collection<values>>.
>
> PTable<Integer, Collection<TupleN>> collectValues =
> classifiedData.collectValues();
> PTable<String, Integer> scores = collectValues.parallelDo("compute
> pairs",
> new PTableScoreCalculator(), Writables.tableOf(Writables.strings(),
> Writables.ints()));
>
>
> Now when I do ParalledDo on the new collection I get back a Pair, having
> keyType and ArrayList<valueType>, over which I can do things as I wish.
>
> class PTableScoreCalculator extends DoFn<Pair<Integer, Collection<TupleN>>,
> Pair<String, Integer>> {
> public void process(Pair<Integer, Collection<TupleN>> input,
>
> Emitter<Pair<String, Integer>> emitter) {
> Iterator<TupleN> primary = input.second().iterator();
> .....................
> }
>
> This way I could iterate over again and again, any comments on the same. I
> am attaching my test case for reference.
>
> BTW why are there two methods that can do the same things the groupBykey
> method and the collectValues method ? I see an Aggregation gets invoked for
> the collection API and in the other case a lazy collection gets created. Any
> idea on the different applications of the two.
>
> regards,
> Rahul
>
>
>
> On 28-06-2012 14:17, Gabriel Reid wrote:
>
> On Thu, Jun 28, 2012 at 9:29 AM, Rahul <rs...@xebia.com> wrote:
>
> Yes indeed this is a small PoC to get familiar with Crunch in relation to my
> problem. Basically I have the following algo at play:
> 1. Read data rows
> 2. Create custom keys for each of them, built using various attributes of
> data (this time it is just a simple hash code, but I would like to emit
> multiple key-value pairs)
> 3. Group similar data based on created Keys
> 4. Iterate over individual items in the group and do extensive comparison
> between all of them
>
> I just built an outline in the test case to see what/how can be done, can
> you advise something better ?
>
> Thanks for the outline. In this case, your approach (with putting the
> contents of the
> incoming Iterable into a collection) should work fine, as long as
> number of elements
> per group is relatively small (i.e. easily able to fit in the memory
> available to each reducer in your Hadoop cluster).
>
> - Gabriel
>
>
--
Director of Data Science
Cloudera
Twitter: @josh_wills
Re: Exception due to same iterator returned back by PGroupedTableType
Posted by Rahul <rs...@xebia.com>.
Hi Gabriel,
I have found a way by which Crunch supports the uses case of having
repeated iterators but I am not completely sure of the in-outs of the same.
Basically rather than doing a groupBy on Ptable to get back a
PGroupedTable, I used the collectValues API to get back a
Ptable<key,Collection<values>>.
/ PTable<Integer, Collection<TupleN>> collectValues =
classifiedData.collectValues();
PTable<String, Integer> scores = collectValues.parallelDo("compute
pairs",
new PTableScoreCalculator(),
Writables.tableOf(Writables.strings(), Writables.ints()));/
Now when I do ParalledDo on the new collection I get back a Pair, having
keyType and ArrayList<valueType>, over which I can do things as I wish.
/class PTableScoreCalculator extends DoFn<Pair<Integer,
Collection<TupleN>>, Pair<String, Integer>> {
//public void process(Pair<Integer, Collection<TupleN>> input,
Emitter<Pair<String, Integer>> emitter) {
Iterator<TupleN> primary = input.second().iterator();
.....................
}/
This way I could iterate over again and again, any comments on the same.
I am attaching my test case for reference.
BTW why are there two methods that can do the same things the
/groupBykey/**method and the /collectValues/ method ? I see an
Aggregation gets invoked for the collection API and in the other case a
lazy collection gets created. Any idea on the different applications of
the two.
regards,
Rahul
On 28-06-2012 14:17, Gabriel Reid wrote:
> On Thu, Jun 28, 2012 at 9:29 AM, Rahul<rs...@xebia.com> wrote:
>> Yes indeed this is a small PoC to get familiar with Crunch in relation to my
>> problem. Basically I have the following algo at play:
>> 1. Read data rows
>> 2. Create custom keys for each of them, built using various attributes of
>> data (this time it is just a simple hash code, but I would like to emit
>> multiple key-value pairs)
>> 3. Group similar data based on created Keys
>> 4. Iterate over individual items in the group and do extensive comparison
>> between all of them
>>
>> I just built an outline in the test case to see what/how can be done, can
>> you advise something better ?
>
> Thanks for the outline. In this case, your approach (with putting the
> contents of the
> incoming Iterable into a collection) should work fine, as long as
> number of elements
> per group is relatively small (i.e. easily able to fit in the memory
> available to each reducer in your Hadoop cluster).
>
> - Gabriel
Re: Exception due to same iterator returned back by PGroupedTableType
Posted by Gabriel Reid <ga...@gmail.com>.
On Thu, Jun 28, 2012 at 9:29 AM, Rahul <rs...@xebia.com> wrote:
> Yes indeed this is a small PoC to get familiar with Crunch in relation to my
> problem. Basically I have the following algo at play:
> 1. Read data rows
> 2. Create custom keys for each of them, built using various attributes of
> data (this time it is just a simple hash code, but I would like to emit
> multiple key-value pairs)
> 3. Group similar data based on created Keys
> 4. Iterate over individual items in the group and do extensive comparison
> between all of them
>
> I just built an outline in the test case to see what/how can be done, can
> you advise something better ?
Thanks for the outline. In this case, your approach (with putting the
contents of the
incoming Iterable into a collection) should work fine, as long as
number of elements
per group is relatively small (i.e. easily able to fit in the memory
available to each reducer in your Hadoop cluster).
- Gabriel
Re: Exception due to same iterator returned back by PGroupedTableType
Posted by Rahul <rs...@xebia.com>.
Hi Gabriel,
Yes indeed this is a small PoC to get familiar with Crunch in relation
to my problem. Basically I have the following algo at play:
1. Read data rows
2. Create custom keys for each of them, built using various attributes
of data (this time it is just a simple hash code, but I would like to
emit multiple key-value pairs)
3. Group similar data based on created Keys
4. Iterate over individual items in the group and do extensive
comparison between all of them
I just built an outline in the test case to see what/how can be done,
can you advise something better ?
regards
Rahul
On 28-06-2012 12:30, Gabriel Reid wrote:
> Hi Rahul,
>
> Ok, looks like I misunderstood your code. In that case, you're indeed
> correct that a
> PeekingIterator won't help you -- it looks like you will indeed need
> to store the data
> in a collection per group in order to do the processing that you're
> trying to do.
>
> Am I correct in assuming that this code is an attempt to get familiar
> with Crunch,
> and less about solving a real-world problem right now? If you are trying to put
> together a solution for a problem, maybe you could outline what you're trying
> to get to -- there may be a better way to get there. I noticed that
> you're grouping
> values by the hash code of the input line, which looks questionable to me.
>
> Regards,
>
> Gabriel
>
> On Thu, Jun 28, 2012 at 8:05 AM, Rahul<rs...@xebia.com> wrote:
>> Hi Gabriel,
>>
>> I am doing n*(n-1) comparisons here every element would be compared with
>> every other element, so peeking iterator would not help much. It would give
>> me the next element but I need to keep all the elements that have been
>> accessed once in another Collection so that I can iterate over them again
>> and again.
>> or Is there some thing that would help here ?
>>
>> regards,
>> Rahul
>>
>>
>> On 27-06-2012 17:48, Gabriel Reid wrote:
>>> On Wed, Jun 27, 2012 at 1:41 PM, Rahul<rs...@xebia.com> wrote:
>>>> I am trying to create multiple iterators in a DoFn process method.
>>>>
>>>> public void process(Pair<Integer, Iterable<TupleN>> input,
>>>> Emitter<Pair<String, Integer>> emitter) {}
>>>>
>>>> Every time I ask a iterator it gives back the same one and thus I could
>>>> not
>>>> not traverse the list again and again as I am hitting the following stack
>>>> trace .
>>> The Iterable.iterator call always returns the same iterator is because
>>> this
>>> is the behaviour that is inherited from the reduce method of the Hadoop
>>> Reducer class (and this behaviour is there because of the underlying way
>>> in which Hadoop MapReduce functions). In both Crunch and pure MapReduce,
>>> you've just got one shot at looping over an Iterable in a reducer (or DoFn
>>> that is functioning on a PGroupedTable).
>>>
>>> If I understood your code correctly, you're trying to loop over an
>>> Iterable
>>> while looking at two consecutive elements at a time. Probably the easiest
>>> way of doing this is using the PeekingIterator class in Google Guava
>>>
>>> (http://docs.guava-libraries.googlecode.com/git-history/release/javadoc/com/google/common/collect/Iterators.html#peekingIterator%28java.util.Iterator%29).
>>> This will allow
>>> you to look one element ahead within an iterator.
>>>
>>> Regards,
>>>
>>> Gabriel
>>
Re: Exception due to same iterator returned back by PGroupedTableType
Posted by Gabriel Reid <ga...@gmail.com>.
Hi Rahul,
Ok, looks like I misunderstood your code. In that case, you're indeed
correct that a
PeekingIterator won't help you -- it looks like you will indeed need
to store the data
in a collection per group in order to do the processing that you're
trying to do.
Am I correct in assuming that this code is an attempt to get familiar
with Crunch,
and less about solving a real-world problem right now? If you are trying to put
together a solution for a problem, maybe you could outline what you're trying
to get to -- there may be a better way to get there. I noticed that
you're grouping
values by the hash code of the input line, which looks questionable to me.
Regards,
Gabriel
On Thu, Jun 28, 2012 at 8:05 AM, Rahul <rs...@xebia.com> wrote:
> Hi Gabriel,
>
> I am doing n*(n-1) comparisons here every element would be compared with
> every other element, so peeking iterator would not help much. It would give
> me the next element but I need to keep all the elements that have been
> accessed once in another Collection so that I can iterate over them again
> and again.
> or Is there some thing that would help here ?
>
> regards,
> Rahul
>
>
> On 27-06-2012 17:48, Gabriel Reid wrote:
>>
>> On Wed, Jun 27, 2012 at 1:41 PM, Rahul<rs...@xebia.com> wrote:
>>>
>>> I am trying to create multiple iterators in a DoFn process method.
>>>
>>> public void process(Pair<Integer, Iterable<TupleN>> input,
>>> Emitter<Pair<String, Integer>> emitter) {}
>>>
>>> Every time I ask a iterator it gives back the same one and thus I could
>>> not
>>> not traverse the list again and again as I am hitting the following stack
>>> trace .
>>
>> The Iterable.iterator call always returns the same iterator is because
>> this
>> is the behaviour that is inherited from the reduce method of the Hadoop
>> Reducer class (and this behaviour is there because of the underlying way
>> in which Hadoop MapReduce functions). In both Crunch and pure MapReduce,
>> you've just got one shot at looping over an Iterable in a reducer (or DoFn
>> that is functioning on a PGroupedTable).
>>
>> If I understood your code correctly, you're trying to loop over an
>> Iterable
>> while looking at two consecutive elements at a time. Probably the easiest
>> way of doing this is using the PeekingIterator class in Google Guava
>>
>> (http://docs.guava-libraries.googlecode.com/git-history/release/javadoc/com/google/common/collect/Iterators.html#peekingIterator%28java.util.Iterator%29).
>> This will allow
>> you to look one element ahead within an iterator.
>>
>> Regards,
>>
>> Gabriel
>
>
Re: Exception due to same iterator returned back by PGroupedTableType
Posted by Rahul <rs...@xebia.com>.
Hi Gabriel,
I am doing n*(n-1) comparisons here every element would be compared with
every other element, so peeking iterator would not help much. It would
give me the next element but I need to keep all the elements that have
been accessed once in another Collection so that I can iterate over them
again and again.
or Is there some thing that would help here ?
regards,
Rahul
On 27-06-2012 17:48, Gabriel Reid wrote:
> On Wed, Jun 27, 2012 at 1:41 PM, Rahul<rs...@xebia.com> wrote:
>> I am trying to create multiple iterators in a DoFn process method.
>>
>> public void process(Pair<Integer, Iterable<TupleN>> input,
>> Emitter<Pair<String, Integer>> emitter) {}
>>
>> Every time I ask a iterator it gives back the same one and thus I could not
>> not traverse the list again and again as I am hitting the following stack
>> trace .
> The Iterable.iterator call always returns the same iterator is because this
> is the behaviour that is inherited from the reduce method of the Hadoop
> Reducer class (and this behaviour is there because of the underlying way
> in which Hadoop MapReduce functions). In both Crunch and pure MapReduce,
> you've just got one shot at looping over an Iterable in a reducer (or DoFn
> that is functioning on a PGroupedTable).
>
> If I understood your code correctly, you're trying to loop over an Iterable
> while looking at two consecutive elements at a time. Probably the easiest
> way of doing this is using the PeekingIterator class in Google Guava
> (http://docs.guava-libraries.googlecode.com/git-history/release/javadoc/com/google/common/collect/Iterators.html#peekingIterator%28java.util.Iterator%29).
> This will allow
> you to look one element ahead within an iterator.
>
> Regards,
>
> Gabriel
Re: Exception due to same iterator returned back by PGroupedTableType
Posted by Gabriel Reid <ga...@gmail.com>.
On Wed, Jun 27, 2012 at 1:41 PM, Rahul <rs...@xebia.com> wrote:
> I am trying to create multiple iterators in a DoFn process method.
>
> public void process(Pair<Integer, Iterable<TupleN>> input,
> Emitter<Pair<String, Integer>> emitter) {}
>
> Every time I ask a iterator it gives back the same one and thus I could not
> not traverse the list again and again as I am hitting the following stack
> trace .
The Iterable.iterator call always returns the same iterator is because this
is the behaviour that is inherited from the reduce method of the Hadoop
Reducer class (and this behaviour is there because of the underlying way
in which Hadoop MapReduce functions). In both Crunch and pure MapReduce,
you've just got one shot at looping over an Iterable in a reducer (or DoFn
that is functioning on a PGroupedTable).
If I understood your code correctly, you're trying to loop over an Iterable
while looking at two consecutive elements at a time. Probably the easiest
way of doing this is using the PeekingIterator class in Google Guava
(http://docs.guava-libraries.googlecode.com/git-history/release/javadoc/com/google/common/collect/Iterators.html#peekingIterator%28java.util.Iterator%29).
This will allow
you to look one element ahead within an iterator.
Regards,
Gabriel