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