@Matthias: Your point of working with a minimal number of clear concepts is desirable to say the least. :)
The reasoning behind the KeyedDatastream is to associate Flink persisted operator state with the keys of the data that produced it, so that stateful computation becomes scalabe in the future. This should not be tied to the GroupedDataStream, especially not if we are removing the option to create groups without windows as proposed on the Wiki by Stephan. On Sat, Jun 27, 2015 at 4:15 PM, Matthias J. Sax < mj...@informatik.hu-berlin.de> wrote: > This was more a conceptual point-of-view argument. From an > implementation point of view, skipping the window building step is a > good idea if a tumbling 1-tuple-size window is detected. > > I prefer to work with a minimum number of concepts (and apply internal > optimization if possible) instead of using redundant concepts for > special cases. Of course, this is my personal point of view. > > -Matthias > > On 06/27/2015 03:47 PM, Aljoscha Krettek wrote: > > What do you mean by Comment 2? Using the whole window apparatus if you > just > > want to have, for example, a simple partitioned filter with partitioned > > state seems a bit extravagant. > > > > On Sat, 27 Jun 2015 at 15:19 Matthias J. Sax < > mj...@informatik.hu-berlin.de> > > wrote: > > > >> Nice starting point. > >> > >> Comment 1: > >> "Each individual stream partition delivers elements strictly in order." > >> (in 'Parallel Streams, Partitions, Time, and Ordering') > >> > >> I would say "FIFO" and not "strictly in order". If data is not emitted > >> in-order, the stream partition will not be in-order either. > >> > >> Comment 2: > >> Why do we need KeyedDataStream. You can get everything done with > >> GroupedDataStream (using a tumbling window of size = 1 tuple). > >> > >> > >> -Matthias > >> > >> On 06/26/2015 07:42 PM, Stephan Ewen wrote: > >>> Here is a first bit of what I have been writing down. Will add more > over > >>> the next days: > >>> > >>> > >>> https://cwiki.apache.org/confluence/display/FLINK/Stream+Windows > >>> > >>> > >> > https://cwiki.apache.org/confluence/display/FLINK/Parallel+Streams%2C+Partitions%2C+Time%2C+and+Ordering > >>> > >>> > >>> > >>> On Thu, Jun 25, 2015 at 6:35 PM, Paris Carbone <par...@kth.se> wrote: > >>> > >>>> +1 for writing this down > >>>> > >>>>> On 25 Jun 2015, at 18:11, Aljoscha Krettek <aljos...@apache.org> > >> wrote: > >>>>> > >>>>> +1 go ahead > >>>>> > >>>>> On Thu, 25 Jun 2015 at 18:02 Stephan Ewen <se...@apache.org> wrote: > >>>>> > >>>>>> Hey! > >>>>>> > >>>>>> This thread covers many different topics. Lets break this up into > >>>> separate > >>>>>> discussions. > >>>>>> > >>>>>> - Operator State is already driven by Gyula and Paris and happening > on > >>>> the > >>>>>> above mentioned pull request and the followup discussions. > >>>>>> > >>>>>> - For windowing, this discussion has brought some results that we > >> should > >>>>>> sum up and clearly write down. > >>>>>> I would like to chime in to do that based on what I learned from > the > >>>>>> document and this discussion. I also got some input from Marton > about > >>>> what > >>>>>> he learned from mapping the Cloud DataFlow constructs to Flink. > >>>>>> I'll draft a Wiki page (with the help of Aljoscha, Marton) that > sums > >>>>>> this up and documents it for users (if we decide to adopt this). > >>>>>> Then we run this by Gyula, Matthias Sax and Kostas for feedback. > >>>>>> > >>>>>> - API style discussions should be yet another thread. This will > >> probably > >>>>>> be opened as people start to address that. > >>>>>> > >>>>>> > >>>>>> I'll try to get a draft of the wiki version out tomorrow noon and > send > >>>> the > >>>>>> link around. > >>>>>> > >>>>>> Greetings, > >>>>>> Stephan > >>>>>> > >>>>>> > >>>>>> > >>>>>> On Thu, Jun 25, 2015 at 3:51 PM, Matthias J. Sax < > >>>>>> mj...@informatik.hu-berlin.de> wrote: > >>>>>> > >>>>>>> Sure. I picked this up. Using the current model for "occurrence > time > >>>>>>> semantics" does not work. > >>>>>>> > >>>>>>> I elaborated on this in the past many times (but nobody cared). It > is > >>>>>>> important to make it clear to the user what semantics are > supported. > >>>>>>> Claiming to support "sliding windows" doesn't mean anything; there > >> are > >>>>>>> too many different semantics out there. :) > >>>>>>> > >>>>>>> > >>>>>>> On 06/25/2015 03:35 PM, Aljoscha Krettek wrote: > >>>>>>>> Yes, I am aware of this requirement and it would also be supported > >> in > >>>>>> my > >>>>>>>> proposed model. > >>>>>>>> > >>>>>>>> The problem is, that the "custom timestamp" feature gives the > >>>>>> impression > >>>>>>>> that the elements would be windowed according to a user-timestamp. > >> The > >>>>>>>> results, however, are wrong because of the assumption about > elements > >>>>>>>> arriving in order. (This is what I was trying to show with my > fancy > >>>>>> ASCII > >>>>>>>> art and result output. > >>>>>>>> > >>>>>>>> On Thu, 25 Jun 2015 at 15:26 Matthias J. Sax < > >>>>>>> mj...@informatik.hu-berlin.de> > >>>>>>>> wrote: > >>>>>>>> > >>>>>>>>> Hi Aljoscha, > >>>>>>>>> > >>>>>>>>> I like that you are pushing in this direction. However, IMHO you > >>>>>>>>> misinterpreter the current approach. It does not assume that > tuples > >>>>>>>>> arrive in-order; the current approach has no notion about a > >>>>>>>>> pre-defined-order (for example, the order in which the event are > >>>>>>>>> created). There is only the notion of "arrival-order" at the > >>>> operator. > >>>>>>>>> From this "arrival-order" perspective, the result are correct(!). > >>>>>>>>> > >>>>>>>>> Windowing in the current approach means for example, "sum up an > >>>>>>>>> attribute of all events you *received* in the last 5 seconds". > That > >>>>>> is a > >>>>>>>>> different meaning that "sum up an attribute of all event that > >>>>>> *occurred* > >>>>>>>>> in the last 5 seconds". Both queries are valid and Flink should > >>>>>> support > >>>>>>>>> both IMHO. > >>>>>>>>> > >>>>>>>>> > >>>>>>>>> -Matthias > >>>>>>>>> > >>>>>>>>> > >>>>>>>>> > >>>>>>>>> On 06/25/2015 03:03 PM, Aljoscha Krettek wrote: > >>>>>>>>>> Yes, now this also processes about 3 mio Elements (Window Size 5 > >>>> sec, > >>>>>>>>> Slide > >>>>>>>>>> 1 sec) but it still fluctuates a lot between 1 mio. and 5 mio. > >>>>>>>>>> > >>>>>>>>>> Performance is not my main concern, however. My concern is that > >> the > >>>>>>>>> current > >>>>>>>>>> model assumes elements to arrive in order, which is simply not > >> true. > >>>>>>>>>> > >>>>>>>>>> In your code you have these lines for specifying the window: > >>>>>>>>>> .window(Time.of(1l, TimeUnit.SECONDS)) > >>>>>>>>>> .every(Time.of(1l, TimeUnit.SECONDS)) > >>>>>>>>>> > >>>>>>>>>> Although this semantically specifies a tumbling window of size 1 > >> sec > >>>>>>> I'm > >>>>>>>>>> afraid it uses the sliding window logic internally (because of > the > >>>>>>>>>> .every()). > >>>>>>>>>> > >>>>>>>>>> In my tests I only have the first line. > >>>>>>>>>> > >>>>>>>>>> > >>>>>>>>>> On Thu, 25 Jun 2015 at 14:32 Gábor Gévay <gga...@gmail.com> > >> wrote: > >>>>>>>>>> > >>>>>>>>>>> I'm very sorry, I had a bug in the InversePreReducer. It should > >> be > >>>>>>>>>>> fixed now. Can you please run it again? > >>>>>>>>>>> > >>>>>>>>>>> I also tried to reproduce some of your performance numbers, but > >> I'm > >>>>>>>>>>> getting only less than 1/10th of yours. For example, in the > >>>> Tumbling > >>>>>>>>>>> case, Current/Reduce produces only ~100000 for me. Do you have > >> any > >>>>>>>>>>> idea what I could be doing wrong? My code: > >>>>>>>>>>> http://pastebin.com/zbEjmGhk > >>>>>>>>>>> I am running it on a 2 GHz Core i7. > >>>>>>>>>>> > >>>>>>>>>>> Best regards, > >>>>>>>>>>> Gabor > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> 2015-06-25 12:31 GMT+02:00 Aljoscha Krettek < > aljos...@apache.org > >>> : > >>>>>>>>>>>> Hi, > >>>>>>>>>>>> I also ran the tests on top of PR 856 (inverse reducer) now. > The > >>>>>>>>> results > >>>>>>>>>>>> seem incorrect. When I insert a Thread.sleep(1) in the tuple > >>>>>> source, > >>>>>>>>> all > >>>>>>>>>>>> the previous tests reported around 3600 tuples (Size 5 sec, > >> Slide > >>>> 1 > >>>>>>>>> sec) > >>>>>>>>>>>> (Theoretically there would be 5000 tuples in 5 seconds but > this > >> is > >>>>>>> due > >>>>>>>>> to > >>>>>>>>>>>> overhead). These are the results for the inverse reduce > >>>>>> optimisation: > >>>>>>>>>>>> (Tuple 0,38) > >>>>>>>>>>>> (Tuple 0,829) > >>>>>>>>>>>> (Tuple 0,1625) > >>>>>>>>>>>> (Tuple 0,2424) > >>>>>>>>>>>> (Tuple 0,3190) > >>>>>>>>>>>> (Tuple 0,3198) > >>>>>>>>>>>> (Tuple 0,-339368) > >>>>>>>>>>>> (Tuple 0,-1315725) > >>>>>>>>>>>> (Tuple 0,-2932932) > >>>>>>>>>>>> (Tuple 0,-5082735) > >>>>>>>>>>>> (Tuple 0,-7743256) > >>>>>>>>>>>> (Tuple 0,75701046) > >>>>>>>>>>>> (Tuple 0,642829470) > >>>>>>>>>>>> (Tuple 0,2242018381) > >>>>>>>>>>>> (Tuple 0,5190708618) > >>>>>>>>>>>> (Tuple 0,10060360311) > >>>>>>>>>>>> (Tuple 0,-94254951) > >>>>>>>>>>>> (Tuple 0,-219806321293) > >>>>>>>>>>>> (Tuple 0,-1258895232699) > >>>>>>>>>>>> (Tuple 0,-4074432596329) > >>>>>>>>>>>> > >>>>>>>>>>>> One line is one emitted window count. This is what happens > when > >> I > >>>>>>>>> remove > >>>>>>>>>>>> the Thread.sleep(1): > >>>>>>>>>>>> (Tuple 0,660676) > >>>>>>>>>>>> (Tuple 0,2553733) > >>>>>>>>>>>> (Tuple 0,3542696) > >>>>>>>>>>>> (Tuple 0,1) > >>>>>>>>>>>> (Tuple 0,1107035) > >>>>>>>>>>>> (Tuple 0,2549491) > >>>>>>>>>>>> (Tuple 0,4100387) > >>>>>>>>>>>> (Tuple 0,-8406583360092) > >>>>>>>>>>>> (Tuple 0,-8406582150743) > >>>>>>>>>>>> (Tuple 0,-8406580427190) > >>>>>>>>>>>> (Tuple 0,-8406580427190) > >>>>>>>>>>>> (Tuple 0,-8406580427190) > >>>>>>>>>>>> (Tuple 0,6847279255682044995) > >>>>>>>>>>>> (Tuple 0,6847279255682044995) > >>>>>>>>>>>> (Tuple 0,-5390528042713628318) > >>>>>>>>>>>> (Tuple 0,-5390528042711551780) > >>>>>>>>>>>> (Tuple 0,-5390528042711551780) > >>>>>>>>>>>> > >>>>>>>>>>>> So at some point the pre-reducer seems to go haywire and does > >> not > >>>>>>>>> recover > >>>>>>>>>>>> from it. The good thing is that it does produce results now, > >> where > >>>>>>> the > >>>>>>>>>>>> previous Current/Reduce would simply hang and not produce any > >>>>>> output. > >>>>>>>>>>>> > >>>>>>>>>>>> On Thu, 25 Jun 2015 at 12:02 Gábor Gévay <gga...@gmail.com> > >>>> wrote: > >>>>>>>>>>>> > >>>>>>>>>>>>> Hello, > >>>>>>>>>>>>> > >>>>>>>>>>>>> Aljoscha, can you please try the performance test of > >>>>>> Current/Reduce > >>>>>>>>>>>>> with the InversePreReducer in PR 856? (If you just call sum, > it > >>>>>> will > >>>>>>>>>>>>> use an InversePreReducer.) It would be an interesting test, > >>>>>> because > >>>>>>>>>>>>> the inverse function optimization really depends on the > stream > >>>>>> being > >>>>>>>>>>>>> ordered, and I think it has the potential of being faster > then > >>>>>>>>>>>>> Next/Reduce. Especially if the window size is much larger > than > >>>> the > >>>>>>>>>>>>> slide size. > >>>>>>>>>>>>> > >>>>>>>>>>>>> Best regards, > >>>>>>>>>>>>> Gabor > >>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>>>> 2015-06-25 11:36 GMT+02:00 Aljoscha Krettek < > >> aljos...@apache.org > >>>>>>> : > >>>>>>>>>>>>>> I think I'll have to elaborate a bit so I created a > >>>>>>> proof-of-concept > >>>>>>>>>>>>>> implementation of my Ideas and ran some throughput > >> measurements > >>>>>> to > >>>>>>>>>>>>>> alleviate concerns about performance. > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> First, though, I want to highlight again why the current > >>>> approach > >>>>>>>>> does > >>>>>>>>>>>>> not > >>>>>>>>>>>>>> work with out-of-order elements (which, again, occur > >> constantly > >>>>>> due > >>>>>>>>> to > >>>>>>>>>>>>> the > >>>>>>>>>>>>>> distributed nature of the system). This is the example I > >> posted > >>>>>>>>>>> earlier: > >>>>>>>>>>>>>> https://gist.github.com/aljoscha/a367012646ab98208d27. The > >> plan > >>>>>>>>> looks > >>>>>>>>>>>>> like > >>>>>>>>>>>>>> this: > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> +--+ > >>>>>>>>>>>>>> | | Source > >>>>>>>>>>>>>> +--+ > >>>>>>>>>>>>>> | > >>>>>>>>>>>>>> +-----+ > >>>>>>>>>>>>>> | | > >>>>>>>>>>>>>> | +--+ > >>>>>>>>>>>>>> | | | Identity Map > >>>>>>>>>>>>>> | +--+ > >>>>>>>>>>>>>> | | > >>>>>>>>>>>>>> +-----+ > >>>>>>>>>>>>>> | > >>>>>>>>>>>>>> +--+ > >>>>>>>>>>>>>> | | Window > >>>>>>>>>>>>>> +--+ > >>>>>>>>>>>>>> | > >>>>>>>>>>>>>> | > >>>>>>>>>>>>>> +--+ > >>>>>>>>>>>>>> | | Sink > >>>>>>>>>>>>>> +--+ > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> So all it does is pass the elements through an identity map > >> and > >>>>>>> then > >>>>>>>>>>>>> merge > >>>>>>>>>>>>>> them again before the window operator. The source emits > >>>> ascending > >>>>>>>>>>>>> integers > >>>>>>>>>>>>>> and the window operator has a custom timestamp extractor > that > >>>>>> uses > >>>>>>>>> the > >>>>>>>>>>>>>> integer itself as the timestamp and should create windows of > >>>>>> size 4 > >>>>>>>>>>> (that > >>>>>>>>>>>>>> is elements with timestamp 0-3 are one window, the next are > >> the > >>>>>>>>>>> elements > >>>>>>>>>>>>>> with timestamp 4-8, and so on). Since the topology basically > >>>>>>> doubles > >>>>>>>>>>> the > >>>>>>>>>>>>>> elements form the source I would expect to get these > windows: > >>>>>>>>>>>>>> Window: 0, 0, 1, 1, 2, 2, 3, 3 > >>>>>>>>>>>>>> Window: 4, 4, 6, 6, 7, 7, 8, 8 > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> The output is this, however: > >>>>>>>>>>>>>> Window: 0, 1, 2, 3, > >>>>>>>>>>>>>> Window: 4, 0, 1, 2, 3, 4, 5, 6, 7, > >>>>>>>>>>>>>> Window: 8, 9, 10, 11, > >>>>>>>>>>>>>> Window: 12, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, > >>>>>>>>>>>>>> Window: 16, 17, 18, 19, > >>>>>>>>>>>>>> Window: 20, 21, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, > >>>>>>>>>>>>>> Window: 24, 25, 26, 27, > >>>>>>>>>>>>>> Window: 28, 29, 30, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> The reason is that the elements simply arrive out-of-order. > >>>>>> Imagine > >>>>>>>>>>> what > >>>>>>>>>>>>>> would happen if the elements actually arrived with some > delay > >>>>>> from > >>>>>>>>>>>>>> different operations. > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> Now, on to the performance numbers. The proof-of-concept I > >>>>>> created > >>>>>>> is > >>>>>>>>>>>>>> available here: > >>>>>>>>>>>>>> > >>>> https://github.com/aljoscha/flink/tree/event-time-window-fn-mock > >>>>>> . > >>>>>>>>> The > >>>>>>>>>>>>> basic > >>>>>>>>>>>>>> idea is that sources assign the current timestamp when > >> emitting > >>>>>>>>>>> elements. > >>>>>>>>>>>>>> They also periodically emit watermarks that tell us that no > >>>>>>> elements > >>>>>>>>>>> with > >>>>>>>>>>>>>> an earlier timestamp will be emitted. The watermarks > propagate > >>>>>>>>> through > >>>>>>>>>>>>> the > >>>>>>>>>>>>>> operators. The window operator looks at the timestamp of an > >>>>>> element > >>>>>>>>>>> and > >>>>>>>>>>>>>> puts it into the buffer that corresponds to that window. > When > >>>> the > >>>>>>>>>>> window > >>>>>>>>>>>>>> operator receives a watermark it will look at the in-flight > >>>>>> windows > >>>>>>>>>>>>>> (basically the buffers) and emit those windows where the > >>>>>> window-end > >>>>>>>>> is > >>>>>>>>>>>>>> before the watermark. > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> For measuring throughput I did the following: The source > emits > >>>>>>> tuples > >>>>>>>>>>> of > >>>>>>>>>>>>>> the form ("tuple", 1) in an infinite loop. The window > operator > >>>>>> sums > >>>>>>>>> up > >>>>>>>>>>>>> the > >>>>>>>>>>>>>> tuples, thereby counting how many tuples the window operator > >> can > >>>>>>>>>>> handle > >>>>>>>>>>>>> in > >>>>>>>>>>>>>> a given time window. There are two different implementations > >> for > >>>>>>> the > >>>>>>>>>>>>>> summation: 1) simply summing up the values in a mapWindow(), > >>>>>> there > >>>>>>>>> you > >>>>>>>>>>>>> get > >>>>>>>>>>>>>> a List of all tuples and simple iterate over it. 2) using > >>>> sum(1), > >>>>>>>>>>> which > >>>>>>>>>>>>> is > >>>>>>>>>>>>>> implemented as a reduce() (that uses the pre-reducer > >>>>>>> optimisations). > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> These are the performance numbers (Current is the current > >>>>>>>>>>> implementation, > >>>>>>>>>>>>>> Next is my proof-of-concept): > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> Tumbling (1 sec): > >>>>>>>>>>>>>> - Current/Map: 1.6 mio > >>>>>>>>>>>>>> - Current/Reduce: 2 mio > >>>>>>>>>>>>>> - Next/Map: 2.2 mio > >>>>>>>>>>>>>> - Next/Reduce: 4 mio > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> Sliding (5 sec, slide 1 sec): > >>>>>>>>>>>>>> - Current/Map: ca 3 mio (fluctuates a lot) > >>>>>>>>>>>>>> - Current/Reduce: No output > >>>>>>>>>>>>>> - Next/Map: ca 4 mio (fluctuates) > >>>>>>>>>>>>>> - Next/Reduce: 10 mio > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> The Next/Reduce variant can basically scale indefinitely > with > >>>>>>> window > >>>>>>>>>>> size > >>>>>>>>>>>>>> because the internal state does not rely on the number of > >>>>>> elements > >>>>>>>>>>> (it is > >>>>>>>>>>>>>> just the current sum). The pre-reducer for sliding elements > >>>>>> cannot > >>>>>>>>>>> handle > >>>>>>>>>>>>>> the amount of tuples, it produces no output. For the two Map > >>>>>>> variants > >>>>>>>>>>> the > >>>>>>>>>>>>>> performance fluctuates because they always keep all the > >> elements > >>>>>> in > >>>>>>>>> an > >>>>>>>>>>>>>> internal buffer before emission, this seems to tax the > garbage > >>>>>>>>>>> collector > >>>>>>>>>>>>> a > >>>>>>>>>>>>>> bit and leads to random pauses. > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> One thing that should be noted is that I had to disable the > >>>>>>>>>>> fake-element > >>>>>>>>>>>>>> emission thread, otherwise the Current versions would > >> deadlock. > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> So, I started working on this because I thought that > >>>> out-of-order > >>>>>>>>>>>>>> processing would be necessary for correctness. And it is > >>>>>> certainly, > >>>>>>>>>>> But > >>>>>>>>>>>>> the > >>>>>>>>>>>>>> proof-of-concept also shows that performance can be greatly > >>>>>>> improved. > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> On Wed, 24 Jun 2015 at 09:46 Gyula Fóra < > gyula.f...@gmail.com > >>> > >>>>>>>>> wrote: > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> I agree lets separate these topics from each other so we > can > >>>> get > >>>>>>>>>>> faster > >>>>>>>>>>>>>>> resolution. > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> There is already a state discussion in the thread we > started > >>>>>> with > >>>>>>>>>>> Paris. > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> On Wed, Jun 24, 2015 at 9:24 AM Kostas Tzoumas < > >>>>>>> ktzou...@apache.org > >>>>>>>>>> > >>>>>>>>>>>>>> wrote: > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> I agree with supporting out-of-order out of the box :-), > >> even > >>>>>> if > >>>>>>>>>>> this > >>>>>>>>>>>>>> means > >>>>>>>>>>>>>>>> a major refactoring. This is the right time to refactor > the > >>>>>>>>>>> streaming > >>>>>>>>>>>>>> API > >>>>>>>>>>>>>>>> before we pull it out of beta. I think that this is more > >>>>>>> important > >>>>>>>>>>>>> than > >>>>>>>>>>>>>> new > >>>>>>>>>>>>>>>> features in the streaming API, which can be prioritized > once > >>>>>> the > >>>>>>>>>>> API > >>>>>>>>>>>>> is > >>>>>>>>>>>>>> out > >>>>>>>>>>>>>>>> of beta (meaning, that IMO this is the right time to stall > >> PRs > >>>>>>>>>>> until > >>>>>>>>>>>>> we > >>>>>>>>>>>>>>>> agree on the design). > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> There are three sections in the document: windowing, > state, > >>>> and > >>>>>>>>>>> API. > >>>>>>>>>>>>> How > >>>>>>>>>>>>>>>> convoluted are those with each other? Can we separate the > >>>>>>>>>>> discussion > >>>>>>>>>>>>> or > >>>>>>>>>>>>>> do > >>>>>>>>>>>>>>>> we need to discuss those all together? I think part of the > >>>>>>>>>>> difficulty > >>>>>>>>>>>>> is > >>>>>>>>>>>>>>>> that we are discussing three design choices at once. > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> On Tue, Jun 23, 2015 at 7:43 PM, Ted Dunning < > >>>>>>>>>>> ted.dunn...@gmail.com> > >>>>>>>>>>>>>>>> wrote: > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> Out of order is ubiquitous in the real-world. Typically, > >>>> what > >>>>>>>>>>>>>> happens is > >>>>>>>>>>>>>>>>> that businesses will declare a maximum allowable delay > for > >>>>>>>>>>> delayed > >>>>>>>>>>>>>>>>> transactions and will commit to results when that delay > is > >>>>>>>>>>> reached. > >>>>>>>>>>>>>>>>> Transactions that arrive later than this cutoff are > >> collected > >>>>>>>>>>>>>> specially > >>>>>>>>>>>>>>>> as > >>>>>>>>>>>>>>>>> corrections which are reported/used when possible. > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> Clearly, ordering can also be violated during processing, > >> but > >>>>>> if > >>>>>>>>>>> the > >>>>>>>>>>>>>> data > >>>>>>>>>>>>>>>>> is originally out of order the situation can't be > repaired > >> by > >>>>>>> any > >>>>>>>>>>>>>>>> protocol > >>>>>>>>>>>>>>>>> fixes that prevent transactions from becoming disordered > >> but > >>>>>> has > >>>>>>>>>>> to > >>>>>>>>>>>>>>>> handled > >>>>>>>>>>>>>>>>> at the data level. > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> On Tue, Jun 23, 2015 at 9:29 AM, Aljoscha Krettek < > >>>>>>>>>>>>> aljos...@apache.org > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> wrote: > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> I also don't like big changes but sometimes they are > >>>>>> necessary. > >>>>>>>>>>>>> The > >>>>>>>>>>>>>>>>> reason > >>>>>>>>>>>>>>>>>> why I'm so adamant about out-of-order processing is that > >>>>>>>>>>>>>> out-of-order > >>>>>>>>>>>>>>>>>> elements are not some exception that occurs once in a > >> while; > >>>>>>>>>>> they > >>>>>>>>>>>>>> occur > >>>>>>>>>>>>>>>>>> constantly in a distributed system. For example, in > this: > >>>>>>>>>>>>>>>>>> https://gist.github.com/aljoscha/a367012646ab98208d27 > the > >>>>>>>>>>>>> resulting > >>>>>>>>>>>>>>>>>> windows > >>>>>>>>>>>>>>>>>> are completely bogus because the current windowing > system > >>>>>>>>>>> assumes > >>>>>>>>>>>>>>>>> elements > >>>>>>>>>>>>>>>>>> to globally arrive in order, which is simply not true. > >> (The > >>>>>>>>>>>>> example > >>>>>>>>>>>>>>>> has a > >>>>>>>>>>>>>>>>>> source that generates increasing integers. Then these > pass > >>>>>>>>>>>>> through a > >>>>>>>>>>>>>>>> map > >>>>>>>>>>>>>>>>>> and are unioned with the original DataStream before a > >> window > >>>>>>>>>>>>>> operator.) > >>>>>>>>>>>>>>>>>> This simulates elements arriving from different > operators > >> at > >>>>>> a > >>>>>>>>>>>>>>>> windowing > >>>>>>>>>>>>>>>>>> operator. The example is also DOP=1, I imagine this to > get > >>>>>>>>>>> worse > >>>>>>>>>>>>>> with > >>>>>>>>>>>>>>>>>> higher DOP. > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> What do you mean by costly? As I said, I have a > >>>>>>>>>>> proof-of-concept > >>>>>>>>>>>>>>>>> windowing > >>>>>>>>>>>>>>>>>> operator that can handle out-or-order elements. This is > an > >>>>>>>>>>> example > >>>>>>>>>>>>>>>> using > >>>>>>>>>>>>>>>>>> the current Flink API: > >>>>>>>>>>>>>>>>>> https://gist.github.com/aljoscha/f8dce0691732e344bbe8. > >>>>>>>>>>>>>>>>>> (It is an infinite source of tuples and a 5 second > window > >>>>>>>>>>> operator > >>>>>>>>>>>>>> that > >>>>>>>>>>>>>>>>>> counts the tuples.) The first problem is that this code > >>>>>>>>>>> deadlocks > >>>>>>>>>>>>>>>> because > >>>>>>>>>>>>>>>>>> of the thread that emits fake elements. If I disable the > >>>> fake > >>>>>>>>>>>>>> element > >>>>>>>>>>>>>>>>> code > >>>>>>>>>>>>>>>>>> it works, but the throughput using my mockup is 4 times > >>>>>> higher > >>>>>>>>>>> . > >>>>>>>>>>>>> The > >>>>>>>>>>>>>>>> gap > >>>>>>>>>>>>>>>>>> widens dramatically if the window size increases. > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> So, it actually increases performance (unless I'm > making a > >>>>>>>>>>> mistake > >>>>>>>>>>>>>> in > >>>>>>>>>>>>>>>> my > >>>>>>>>>>>>>>>>>> explorations) and can handle elements that arrive > >>>>>> out-of-order > >>>>>>>>>>>>>> (which > >>>>>>>>>>>>>>>>>> happens basically always in a real-world windowing > >>>>>> use-cases). > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> On Tue, 23 Jun 2015 at 12:51 Stephan Ewen < > >> se...@apache.org > >>>>> > >>>>>>>>>>>>> wrote: > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> What I like a lot about Aljoscha's proposed design is > >> that > >>>>>> we > >>>>>>>>>>>>>> need no > >>>>>>>>>>>>>>>>>>> different code for "system time" vs. "event time". It > >> only > >>>>>>>>>>>>>> differs in > >>>>>>>>>>>>>>>>>> where > >>>>>>>>>>>>>>>>>>> the timestamps are assigned. > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> The OOP approach also gives you the semantics of total > >>>>>>>>>>> ordering > >>>>>>>>>>>>>>>> without > >>>>>>>>>>>>>>>>>>> imposing merges on the streams. > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> On Tue, Jun 23, 2015 at 10:43 AM, Matthias J. Sax < > >>>>>>>>>>>>>>>>>>> mj...@informatik.hu-berlin.de> wrote: > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> I agree that there should be multiple alternatives the > >>>>>>>>>>> user(!) > >>>>>>>>>>>>>> can > >>>>>>>>>>>>>>>>>>>> choose from. Partial out-of-order processing works for > >>>>>>>>>>>>> many/most > >>>>>>>>>>>>>>>>>>>> aggregates. However, if you consider > >>>>>>>>>>> Event-Pattern-Matching, > >>>>>>>>>>>>>> global > >>>>>>>>>>>>>>>>>>>> ordering in necessary (even if the performance penalty > >>>>>>>>>>> might > >>>>>>>>>>>>> be > >>>>>>>>>>>>>>>>> high). > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> I would also keep "system-time windows" as an > >> alternative > >>>>>>>>>>> to > >>>>>>>>>>>>>>>> "source > >>>>>>>>>>>>>>>>>>>> assigned ts-windows". > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> It might also be interesting to consider the following > >>>>>>>>>>> paper > >>>>>>>>>>>>> for > >>>>>>>>>>>>>>>>>>>> overlapping windows: "Resource sharing in continuous > >>>>>>>>>>>>>> sliding-window > >>>>>>>>>>>>>>>>>>>> aggregates" > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>> https://dl.acm.org/citation.cfm?id=1316720 > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> -Matthias > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> On 06/23/2015 10:37 AM, Gyula Fóra wrote: > >>>>>>>>>>>>>>>>>>>>> Hey > >>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>> I think we should not block PRs unnecessarily if your > >>>>>>>>>>>>>> suggested > >>>>>>>>>>>>>>>>>> changes > >>>>>>>>>>>>>>>>>>>>> might touch them at some point. > >>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>> Also I still think we should not put everything in > the > >>>>>>>>>>>>>> Datastream > >>>>>>>>>>>>>>>>>>> because > >>>>>>>>>>>>>>>>>>>>> it will be a huge mess. > >>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>> Also we need to agree on the out of order processing, > >>>>>>>>>>>>> whether > >>>>>>>>>>>>>> we > >>>>>>>>>>>>>>>>> want > >>>>>>>>>>>>>>>>>>> it > >>>>>>>>>>>>>>>>>>>>> the way you proposed it(which is quite costly). > Another > >>>>>>>>>>>>>>>> alternative > >>>>>>>>>>>>>>>>>>>>> approach there which fits in the current windowing is > >> to > >>>>>>>>>>>>>> filter > >>>>>>>>>>>>>>>> out > >>>>>>>>>>>>>>>>>> if > >>>>>>>>>>>>>>>>>>>>> order events and apply a special handling operator on > >>>>>>>>>>> them. > >>>>>>>>>>>>>> This > >>>>>>>>>>>>>>>>>> would > >>>>>>>>>>>>>>>>>>> be > >>>>>>>>>>>>>>>>>>>>> fairly lightweight. > >>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>> My point is that we need to consider some alternative > >>>>>>>>>>>>>> solutions. > >>>>>>>>>>>>>>>>> And > >>>>>>>>>>>>>>>>>> we > >>>>>>>>>>>>>>>>>>>>> should not block contributions along the way. > >>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>> Cheers > >>>>>>>>>>>>>>>>>>>>> Gyula > >>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>> On Tue, Jun 23, 2015 at 9:55 AM Aljoscha Krettek < > >>>>>>>>>>>>>>>>>> aljos...@apache.org> > >>>>>>>>>>>>>>>>>>>>> wrote: > >>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>>> The reason I posted this now is that we need to > think > >>>>>>>>>>> about > >>>>>>>>>>>>>> the > >>>>>>>>>>>>>>>>> API > >>>>>>>>>>>>>>>>>>> and > >>>>>>>>>>>>>>>>>>>>>> windowing before proceeding with the PRs of Gabor > >>>>>>>>>>> (inverse > >>>>>>>>>>>>>>>> reduce) > >>>>>>>>>>>>>>>>>> and > >>>>>>>>>>>>>>>>>>>>>> Gyula (removal of "aggregate" functions on > >> DataStream). > >>>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>>> For the windowing, I think that the current model > does > >>>>>>>>>>> not > >>>>>>>>>>>>>> work > >>>>>>>>>>>>>>>>> for > >>>>>>>>>>>>>>>>>>>>>> out-of-order processing. Therefore, the whole > >> windowing > >>>>>>>>>>>>>>>>>> infrastructure > >>>>>>>>>>>>>>>>>>>> will > >>>>>>>>>>>>>>>>>>>>>> basically have to be redone. Meaning also that any > >> work > >>>>>>>>>>> on > >>>>>>>>>>>>>> the > >>>>>>>>>>>>>>>>>>>>>> pre-aggregators or optimizations that we do now > >> becomes > >>>>>>>>>>>>>> useless. > >>>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>>> For the API, I proposed to restructure the > >> interactions > >>>>>>>>>>>>>> between > >>>>>>>>>>>>>>>>> all > >>>>>>>>>>>>>>>>>>> the > >>>>>>>>>>>>>>>>>>>>>> different *DataStream classes and > grouping/windowing. > >>>>>>>>>>> (See > >>>>>>>>>>>>>> API > >>>>>>>>>>>>>>>>>> section > >>>>>>>>>>>>>>>>>>>> of > >>>>>>>>>>>>>>>>>>>>>> the doc I posted.) > >>>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>>> On Mon, 22 Jun 2015 at 21:56 Gyula Fóra < > >>>>>>>>>>>>> gyula.f...@gmail.com > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> wrote: > >>>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>>>> Hi Aljoscha, > >>>>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>>>> Thanks for the nice summary, this is a very good > >>>>>>>>>>>>> initiative. > >>>>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>>>> I added some comments to the respective sections > >>>>>>>>>>> (where I > >>>>>>>>>>>>>> didnt > >>>>>>>>>>>>>>>>>> fully > >>>>>>>>>>>>>>>>>>>>>> agree > >>>>>>>>>>>>>>>>>>>>>>> :).). > >>>>>>>>>>>>>>>>>>>>>>> At some point I think it would be good to have a > >> public > >>>>>>>>>>>>>> hangout > >>>>>>>>>>>>>>>>>>> session > >>>>>>>>>>>>>>>>>>>>>> on > >>>>>>>>>>>>>>>>>>>>>>> this, which could make a more dynamic discussion. > >>>>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>>>> Cheers, > >>>>>>>>>>>>>>>>>>>>>>> Gyula > >>>>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>>>> Aljoscha Krettek <aljos...@apache.org> ezt írta > >>>>>>>>>>> (időpont: > >>>>>>>>>>>>>>>> 2015. > >>>>>>>>>>>>>>>>>> jún. > >>>>>>>>>>>>>>>>>>>>>> 22., > >>>>>>>>>>>>>>>>>>>>>>> H, 21:34): > >>>>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>>>>> Hi, > >>>>>>>>>>>>>>>>>>>>>>>> with people proposing changes to the streaming > part > >> I > >>>>>>>>>>>>> also > >>>>>>>>>>>>>>>>> wanted > >>>>>>>>>>>>>>>>>> to > >>>>>>>>>>>>>>>>>>>>>>> throw > >>>>>>>>>>>>>>>>>>>>>>>> my hat into the ring. :D > >>>>>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>>>>> During the last few months, while I was getting > >>>>>>>>>>>>> acquainted > >>>>>>>>>>>>>>>> with > >>>>>>>>>>>>>>>>>> the > >>>>>>>>>>>>>>>>>>>>>>>> streaming system, I wrote down some thoughts I had > >>>>>>>>>>> about > >>>>>>>>>>>>>> how > >>>>>>>>>>>>>>>>>> things > >>>>>>>>>>>>>>>>>>>>>> could > >>>>>>>>>>>>>>>>>>>>>>>> be improved. Hopefully, they are in somewhat > >> coherent > >>>>>>>>>>>>> shape > >>>>>>>>>>>>>>>> now, > >>>>>>>>>>>>>>>>>> so > >>>>>>>>>>>>>>>>>>>>>>> please > >>>>>>>>>>>>>>>>>>>>>>>> have a look if you are interested in this: > >>>>>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>> > >>>>>>> > >>>>>> > >>>> > >> > https://docs.google.com/document/d/1rSoHyhUhm2IE30o5tkR8GEetjFvMRMNxvsCfoPsW6_4/edit?usp=sharing > >>>>>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>>>>> This mostly covers: > >>>>>>>>>>>>>>>>>>>>>>>> - Timestamps assigned at sources > >>>>>>>>>>>>>>>>>>>>>>>> - Out-of-order processing of elements in window > >>>>>>>>>>>>> operators > >>>>>>>>>>>>>>>>>>>>>>>> - API design > >>>>>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>>>>> Please let me know what you think. Comment in the > >>>>>>>>>>>>> document > >>>>>>>>>>>>>> or > >>>>>>>>>>>>>>>>> here > >>>>>>>>>>>>>>>>>>> in > >>>>>>>>>>>>>>>>>>>>>> the > >>>>>>>>>>>>>>>>>>>>>>>> mailing list. > >>>>>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>>>>> I have a PR in the makings that would introduce > >> source > >>>>>>>>>>>>>>>>> timestamps > >>>>>>>>>>>>>>>>>>> and > >>>>>>>>>>>>>>>>>>>>>>>> watermarks for keeping track of them. I also > hacked > >> a > >>>>>>>>>>>>>>>>>>> proof-of-concept > >>>>>>>>>>>>>>>>>>>>>>> of a > >>>>>>>>>>>>>>>>>>>>>>>> windowing system that is able to process > >> out-of-order > >>>>>>>>>>>>>> elements > >>>>>>>>>>>>>>>>>>> using a > >>>>>>>>>>>>>>>>>>>>>>>> FlatMap operator. (It uses panes to perform > >> efficient > >>>>>>>>>>>>>>>>>>>>>> pre-aggregations.) > >>>>>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>>>>> Cheers, > >>>>>>>>>>>>>>>>>>>>>>>> Aljoscha > >>>>>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>> > >>>>>>>>> > >>>>>>>>> > >>>>>>>> > >>>>>>> > >>>>>>> > >>>>>> > >>>> > >>>> > >>> > >> > >> > > > >
