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 >>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>> >>>>>>>>> >>>>>>>> >>>>>>> >>>>>>> >>>>>> >>>>> >>>>> >>>> >> >> >
signature.asc
Description: OpenPGP digital signature
