So, would it be a reasonable solution to just modify the WindowUtils.isParallelPolicy() method to return false in case of "eviction instanceof CountEvictionPolicy && trigger instanceof TimeTriggerPolicy" ?
2015-04-30 12:21 GMT+02:00 Gyula Fóra <gyula.f...@gmail.com>: > I'm referring to Peter's problem. If we just create more count discretizers > it doesn't really break the semantics given the network guarantees but it > is not very intuitive. > > On Thursday, April 30, 2015, Aljoscha Krettek <aljos...@apache.org> wrote: > > > @Gyula: Are you referring to the pre-aggregator or the thing Peter > > mentioned? > > > > On Thu, Apr 30, 2015 at 11:12 AM, Gyula Fóra <gyula.f...@gmail.com > > <javascript:;>> wrote: > > > The problem is in the WindowUtils.isParallel policy method. It makes > > count > > > policies automatically parallel as well. > > > > > > On Thursday, April 30, 2015, Aljoscha Krettek <aljos...@apache.org > > <javascript:;>> wrote: > > > > > >> Hi, > > >> no, I think the two are unrelated. But that's another problem we need > > >> to tackle then. > > >> > > >> Cheers, > > >> Aljoscha > > >> > > >> On Thu, Apr 30, 2015 at 9:15 AM, Szabó Péter < > nemderogator...@gmail.com > > <javascript:;> > > >> <javascript:;>> wrote: > > >> > Hey, > > >> > > > >> > our intern, Pablo pointed out that there is some problem with mixed > > >> > windowing policies. When you write > > >> > ... > > >> > .window(Count ...) > > >> > .every(Time ...) > > >> > .mapWindow(...) > > >> > ... > > >> > > > >> > The result makes no sense, as the window is not of the specified > > length. > > >> > Maybe, there is some conflict between Time and Count logic. Is this > a > > >> > related problem to yours? > > >> > > > >> > Peter > > >> > > > >> > 2015-04-29 21:15 GMT+02:00 Aljoscha Krettek <aljos...@apache.org > > <javascript:;> > > >> <javascript:;>>: > > >> > > > >> >> I am on it, yes. :D > > >> >> On Apr 29, 2015 8:31 PM, "Gyula Fóra" <gyf...@apache.org > > <javascript:;> <javascript:;>> > > >> wrote: > > >> >> > > >> >> > Hey, > > >> >> > You are right, so there seems to be some error in the logic that > > >> checks > > >> >> for > > >> >> > the end of the slide. > > >> >> > I checked the count pre-reducer and the error there is clear, but > > if I > > >> >> fix > > >> >> > it it breaks some tests so I think some other stuff must not work > > >> >> properly > > >> >> > either. > > >> >> > > > >> >> > I cannot work on this until I get back from holiday, it would be > > good > > >> if > > >> >> > you could look into it. > > >> >> > > > >> >> > Cheers, > > >> >> > Gyula > > >> >> > > > >> >> > On Wed, Apr 29, 2015 at 7:09 PM, Aljoscha Krettek < > > >> aljos...@apache.org <javascript:;> <javascript:;>> > > >> >> > wrote: > > >> >> > > > >> >> > > Oh, I saw them, but in updateCurrent() (both grouped and > > >> non-grouped) > > >> >> > > the else path is never taken when I run the windowed WordCount > > >> >> > > example. > > >> >> > > > > >> >> > > I have a branch where I just throw a RuntimeException in the > else > > >> >> > > branches and the whole things still works: mvn verify doesn't > > fail, > > >> >> > > the examples still run. When looking at the behaviour of the > > >> >> > > functions, it seems clear to me that the else path can never be > > >> taken. > > >> >> > > > > >> >> > > On Wed, Apr 29, 2015 at 6:15 PM, Gyula Fóra <gyf...@apache.org > > <javascript:;> > > >> <javascript:;>> wrote: > > >> >> > > > Hey, > > >> >> > > > > > >> >> > > > They actually work :P Although I have to admit I need to do > > some > > >> >> > > > refactoring of the method names and parameters. > > >> >> > > > > > >> >> > > > I made some quick refactoring and added some comments for the > > key > > >> >> > > methods: > > >> >> > > > > > >> >> > > > > > >> >> > > > > >> >> > > > >> >> > > >> > > > https://github.com/mbalassi/flink/blob/dfa028ce5102b1f0d098eaec6c4b4d3744bee379/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/SlidingPreReducer.java#L96 > > >> >> > > > > > >> >> > > > > > >> >> > > > > >> >> > > > >> >> > > >> > > > https://github.com/mbalassi/flink/blob/dfa028ce5102b1f0d098eaec6c4b4d3744bee379/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/SlidingPreReducer.java#L105 > > >> >> > > > > > >> >> > > > > > >> >> > > > > >> >> > > > >> >> > > >> > > > https://github.com/mbalassi/flink/blob/dfa028ce5102b1f0d098eaec6c4b4d3744bee379/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/SlidingPreReducer.java#L132 > > >> >> > > > > > >> >> > > > Those three parts contain the key logic, and now the method > > names > > >> >> > should > > >> >> > > > make more sense as well :) > > >> >> > > > > > >> >> > > > Cheers, > > >> >> > > > Gyula > > >> >> > > > > > >> >> > > > > > >> >> > > > > > >> >> > > > > > >> >> > > > > > >> >> > > > On Wed, Apr 29, 2015 at 4:44 PM, Aljoscha Krettek < > > >> >> aljos...@apache.org <javascript:;> <javascript:;> > > >> >> > > > > >> >> > > > wrote: > > >> >> > > > > > >> >> > > >> Hi, > > >> >> > > >> one of my notorious (by now) investigative questions: How do > > >> these > > >> >> > > >> work? To meet it seems that they do not perform any actual > pre > > >> >> > > >> aggregation but just keep all the elements in a LinkedList > > >> >> > > >> (non-grouped case) or HashMap (grouped-case) and aggregate > the > > >> >> > > >> elements when the window is to be emitted. > > >> >> > > >> > > >> >> > > >> Am I missing something again? :D > > >> >> > > >> > > >> >> > > >> Cheers, > > >> >> > > >> Aljoscha > > >> >> > > >> > > >> >> > > > > >> >> > > > >> >> > > >> > > >
