I agree that there are many things that needs to be figured out properly for iterations, and I am okay with postponing them for the next release if we want to get this one out quickly.
The only problem is that this probably breaks the SAMOA connector. Paris can you confirm this? Stephan Ewen <se...@apache.org> ezt írta (időpont: 2015. okt. 8., Cs, 17:12): > For me as an outsider to the iterations, I would say that both approaches > are in some way tricky with some unexpected behavior. > > Parallelism implicitly from the predecessor (input) or the successor (head > task - what happens if there are multiple with different parallelism?) can > confuse in either way. > I have the feeling that what each one perceives as more consistent or > intuitive depends a bit on their mental model of the iterations (given > their prior experience and expectations). > > I agree that we should do something there. But given that we are apparently > not really close to knowing what would be best way to go (or agreeing on > it), I would like to not block 0.10 on this (workarounds are available > after all) and take this for the next release with enough time properly > figure this out and discuss it. > > The iterations will anyways need some work for the next release to > integrate them with checkpointing and watermarks, so would you agree that > we tackle this then as part of an effort to advance the iteration feature > as a whole? > > Greetings, > Stephan > > > > On Thu, Oct 8, 2015 at 4:42 PM, Gyula Fóra <gyula.f...@gmail.com> wrote: > > > The feedback tuples might get rebalanced but the normal input should not. > > > > But still the main problem is the fact that partitioning is not handled > > transparently, and actually does not work when you set the way you > expect. > > > > Gyula > > > > Aljoscha Krettek <aljos...@apache.org> ezt írta (időpont: 2015. okt. 8., > > Cs, 16:33): > > > > > Ok, I see your point. But I think there will be problems no matter what > > > parallelism is chosen for the iteration source/sink. If the parallelism > > of > > > the head is chosen then there will be an implicit rebalance from the > > > operation right before the iteration to the iteration head. I think > this > > > should break ordering as well, in your case. > > > > > > On Tue, 6 Oct 2015 at 10:39 Gyula Fóra <gyula.f...@gmail.com> wrote: > > > > > > > Hi, > > > > > > > > This is just a workaround, which actually breaks input order from my > > > > source. I think the iteration construction should be reworked to set > > the > > > > parallelism of the source/sink to the parallelism of the head > operator > > > (and > > > > validate that all heads have the same parallelism). > > > > > > > > I thought this was the solution that you described with Stephan in > some > > > > older discussion before the rewrite. > > > > > > > > Cheers, > > > > Gyula > > > > > > > > Aljoscha Krettek <aljos...@apache.org> ezt írta (időpont: 2015. okt. > > 6., > > > > K, > > > > 9:15): > > > > > > > > > Hi, > > > > > I think what you would like to to can be achieved by: > > > > > > > > > > IterativeStream it = > in.map(IdentityMap).setParallelism(2).iterate() > > > > > DataStream mapped = it.map(...) > > > > > it.closeWith(mapped.partitionByHash(someField)) > > > > > > > > > > The input is rebalanced to the map inside the iteration as in your > > > > example > > > > > and the feedback should be partitioned by hash. > > > > > > > > > > Cheers, > > > > > Aljoscha > > > > > > > > > > > > > > > On Tue, 6 Oct 2015 at 00:11 Gyula Fóra <gyula.f...@gmail.com> > wrote: > > > > > > > > > > > Hey, > > > > > > > > > > > > This question is mainly targeted towards Aljoscha but maybe > someone > > > can > > > > > > help me out here: > > > > > > > > > > > > I think the way feedback partitioning is handled does not work, > let > > > me > > > > > > illustrate with a simple example: > > > > > > > > > > > > IterativeStream it = ... (parallelism 1) > > > > > > DataStream mapped = it.map(...) (parallelism 2) > > > > > > // this does not work as the feedback has parallelism 2 != 1 > > > > > > // it.closeWith(mapped.partitionByHash(someField)) > > > > > > // so we need rebalance the data > > > > > > > > > > > > > > > > > > > > > > > > > > > it.closeWith(mapped.map(NoOpMap).setParallelism(1).partitionByHash(someField)) > > > > > > > > > > > > This program will execute but the feedback will not be > partitioned > > by > > > > > hash > > > > > > to the mapper instances: > > > > > > The partitioning will be set from the noOpMap to the iteration > sink > > > > which > > > > > > has parallelism different from the mapper (1 vs 2) and then the > > > > iteration > > > > > > source forwards the element to the mapper (always to 0). > > > > > > > > > > > > So the problem is basically that the iteration source/sink pair > > gets > > > > the > > > > > > parallelism of the input stream (p=1) not the head operator (p = > 2) > > > > which > > > > > > leads to incorrect partitioning. > > > > > > > > > > > > Did I miss something here? > > > > > > > > > > > > Cheers, > > > > > > Gyula > > > > > > > > > > > > > > > > > > > > > > > > > > >
