Thanks Leah. Overall LGTM. A few nits:
- the first figure shows window [9,19] but the window is not aligned properly (it should be 1ms to the right; right now, it's aligned to window [8,18]) - in the second figure, a hopping window would create more windows, ie, the first window would be [-6,14) and the last window would be [19,29), thus it's not just 10 windows but 26 windows (if I did not miss count) - "Two new windows will be created by the late record" late -> out-of-order -Matthias On 7/28/20 4:34 PM, Sophie Blee-Goldman wrote: > Thanks for the update Leah -- I think that all makes sense. > > Cheers, > Sophie > > On Tue, Jul 28, 2020 at 3:55 PM Leah Thomas <ltho...@confluent.io> wrote: > >> Another minor tweak, instead of defining the window by the *size*, it will >> be defined by *timeDifference*, which is the maximum time difference >> between two events. This is a more precise way to define a window due to >> its inclusive ends, while allowing the user to create the window they >> expect. This definition fits with the current examples, where a record at >> *10* would fall into a window of time difference *5* from *[5,10]*. This >> window contains any records at 5, 6, 7, 8, 9, and 10, which is 6 instances >> instead of 5. This semantic difference is why I've shifted *size* to >> *timeDifference*. >> >> The new builder will be *withTimeDifferenceAndGrace*, keeping with other >> conventions. >> >> Let me know if there are any concerns! The vote thread is open as well >> here: >> http://mail-archives.apache.org/mod_mbox/kafka-dev/202007.mbox/browser >> >> Best, >> Leah >> >> On Mon, Jul 27, 2020 at 3:58 PM Leah Thomas <ltho...@confluent.io> wrote: >> >>> A small tweak - to make it more clear to users that grace is required, as >>> well as cleaning up some of the confusing grammar semantics of windows, >> the >>> main builder method for *slidingWindows* will be *withSizeAndGrace* >> instead >>> of *of*. This looks like it'll be the last change (for now) on the >>> public API. If anyone has any comments or thoughts let me know. >> Otherwise, >>> I'll take this to vote shortly. >>> >>> Best, >>> Leah >>> >>> On Fri, Jul 24, 2020 at 3:45 PM Leah Thomas <ltho...@confluent.io> >> wrote: >>> >>>> To accommodate the change to a final class, I've added another >>>> *windowedBy()* function in *CogroupedKStream.java *to handle >>>> SlidingWindows. >>>> >>>> As far as the discussion goes, I think this is the last change we've >>>> talked about. If anyone has other comments or concerns, please let me >> know! >>>> >>>> Cheers, >>>> Leah >>>> >>>> On Thu, Jul 23, 2020 at 7:34 PM Leah Thomas <ltho...@confluent.io> >> wrote: >>>> >>>>> Thanks for the discussion about extending TimeWindows. I agree that >>>>> making it future proof is important, and the implementation of >>>>> SlidingWindows is unique enough that it seems logical to make it its >> own >>>>> final class. >>>>> >>>>> On that note, I've updated the KIP to make SlidingWindows a stand alone >>>>> final class, and added the *windowedBy() *API in *KGroupedStream *to >>>>> handle SlidingWindows. It seems that SlidingWindows would still be >> able to >>>>> leverage *TimeWindowedKStream by* creating a SlidingWindows version of >>>>> *TimeWindowedKStreamImpl* that implements *TimeWindowedKStream. *If >>>>> anyone sees issues with this implementation, please let me know. >>>>> >>>>> Best, >>>>> Leah >>>>> >>>>> On Wed, Jul 22, 2020 at 10:47 PM John Roesler <vvcep...@apache.org> >>>>> wrote: >>>>> >>>>>> Thanks for the reply, Sophie. >>>>>> >>>>>> That all sounds about right to me. >>>>>> >>>>>> The Windows “interface”/algorithm is quite flexible, so it makes sense >>>>>> for it to be extensible. Different implementations can (and do) >> enumerate >>>>>> different windows to suit different use cases. >>>>>> >>>>>> On the other hand, I can’t think of any way to extend SessionWindows >> to >>>>>> do something different using the same algorithm, so it makes sense >> for it >>>>>> to stay final. >>>>>> >>>>>> If we think SlidingWindows is similarly not usefully extensible, then >>>>>> we can make it final. It’s easy to remove final later, but adding it >> is not >>>>>> possible. Or we could go the other route and just make it an >> interface, on >>>>>> general principle. Both of these choices are safe API design. >>>>>> >>>>>> Thanks again, >>>>>> John >>>>>> >>>>>> On Wed, Jul 22, 2020, at 21:54, Sophie Blee-Goldman wrote: >>>>>>>> >>>>>>>> Users could pass in a custom `SessionWindows` as >>>>>>>> long as the session algorithm works correctly for it. >>>>>>> >>>>>>> >>>>>>> Well not really, SessionWindows is a final class. TimeWindows is >> also >>>>>> a >>>>>>> final class, so neither of these can be extended/customized. For a >>>>>> given >>>>>>> Windows then there would only be three (non-overlapping) >>>>>> possibilities: >>>>>>> either it's TimeWindows, SlidingWindows, or a custom Windows. I >> don't >>>>>>> think there's any problem with determining what the user wants in >>>>>> this case >>>>>>> -- >>>>>>> we would just check if it's a SlidingWindows and connect the new >>>>>> processor, >>>>>>> or else connect the existing hopping/tumbling window processor. >>>>>>> >>>>>>> I'll admit that last sentence does leave a bad taste in my mouth. >>>>>> Part of >>>>>>> that >>>>>>> is probably the "leaking" API Matthias pointed out; we just assume >> the >>>>>>> hopping/tumbling window implementation fits all custom windows. But >> I >>>>>> guess >>>>>>> if you really needed to customize the algorithm any further you may >>>>>> as well >>>>>>> stick in a transformer and do it all yourself. >>>>>>> >>>>>>> Anyways, given what we have, it does seem weird to apply one >> algorithm >>>>>>> for most Windows types and then swap in a different one for one >>>>>> specific >>>>>>> extension of Windows. So adding a new #windowedBy(SlidingWindows) >>>>>>> sounds reasonable to me. >>>>>>> >>>>>>> I'm still not convinced that we need a whole new TimeWindowedKStream >>>>>>> equivalent class for sliding windows though. It seems like the >>>>>>> hopping/tumbling >>>>>>> window implementation could benefit just as much from a subtractor >> as >>>>>> the >>>>>>> sliding windows so the only future-proofing we get is the ability to >>>>>> be >>>>>>> lazy and >>>>>>> add the subtractor to one but not the other. Of course it would only >>>>>> be an >>>>>>> optimization so we could just not apply it to one type and nothing >>>>>> would >>>>>>> break. >>>>>>> It does make me nervous to go against the "future-proof" direction, >>>>>> though. >>>>>>> Are there any other examples of things we might want to add to one >>>>>> window >>>>>>> type but not to another? >>>>>>> >>>>>>> On another note, this actually brings up a new question: should >>>>>>> SlidingWindows >>>>>>> also be final? My take is "yes" since there's no reasonable >>>>>> customization of >>>>>>> sliding windows, at least not that I can think of. Thoughts? >>>>>>> >>>>>>> >>>>>>> On Wed, Jul 22, 2020 at 7:15 PM John Roesler <vvcep...@apache.org> >>>>>> wrote: >>>>>>> >>>>>>>> Thanks, all, >>>>>>>> >>>>>>>> I can see how my conclusion was kind of a leap. >>>>>>>> >>>>>>>> What Matthias said is indeed what I was thinking. When you >> provide a >>>>>>>> window definition to the windowBy() method, you are selecting an >>>>>> algorithm >>>>>>>> that will be used to compute the windows from the input data. >>>>>>>> >>>>>>>> I didn’t mean the code implementation “algorithm”, but the >>>>>> high-level >>>>>>>> algorithm that describes how the input stream will be transformed >>>>>> into a >>>>>>>> sequence of windows. >>>>>>>> >>>>>>>> For example, the algorithm for Windows is something like “given >> the >>>>>> record >>>>>>>> timestamp, include the record in each of the enumerated windows”. >>>>>> Note that >>>>>>>> there can be a lot of variation in how the windows are enumerated, >>>>>> which is >>>>>>>> why there are at least a couple of implementations of Windows, and >>>>>> we are >>>>>>>> open to adding more (like for natural calendar boundaries). >>>>>>>> >>>>>>>> For SessionWindows, it’s more like “if any window is within the >> gap, >>>>>>>> extend its boundaries to include this record and if two windows >>>>>> touch, then >>>>>>>> merge them”. >>>>>>>> >>>>>>>> Clearly, the algorithm for SlidingWindows doesn’t fall into either >>>>>>>> category, so it seems inappropriate to claim that it does in the >>>>>> API (by >>>>>>>> implementing Windows with stubbed methods) and then cast >> internally >>>>>> to >>>>>>>> execute a completely different algorithm. >>>>>>>> >>>>>>>> To your other observation, that the DSL object resulting from >>>>>> windowBy >>>>>>>> would look the same for Windows and SessionWindows, maybe it makes >>>>>> sense >>>>>>>> for windowBy(SessionWindows) also to return a TimeWindowedKStream. >>>>>>>> >>>>>>>> i.e.: >>>>>>>> ======================= >>>>>>>> <W extends Window> TimeWindowedKStream<K, V> windowedBy(final >>>>>> Windows<W> >>>>>>>> windows); >>>>>>>> TimeWindowedKStream<K, V> windowedBy(final SlidingWindows >> windows); >>>>>>>> ======================= >>>>>>>> >>>>>>>> I can’t think of a reason this wouldn’t work. But then again, it >>>>>> would be >>>>>>>> more future-proof to go ahead and specify a different return type >>>>>> now, if >>>>>>>> we think we'll want to add subtractors and stuff later. I don't >>>>>> have a >>>>>>>> strong feeling about that part of the API. It seems to be >>>>>> independent of >>>>>>>> whether SlidingWindows extends Windows or not. >>>>>>>> >>>>>>>> Thanks, >>>>>>>> -John >>>>>>>> >>>>>>>> On Wed, Jul 22, 2020, at 19:41, Matthias J. Sax wrote: >>>>>>>>> I think what John tries to say is the following: >>>>>>>>> >>>>>>>>> We have `windowedBy(Windows)` that accept hopping/tumbling >>>>>> windows but >>>>>>>>> also custom window and we use a specific algorithm. Note, that >>>>>> custom >>>>>>>>> windows must "work" based on the used algorithm. >>>>>>>>> >>>>>>>>> For session windows we have `windowedBy(SessionWindows)` and >>>>>> apply a >>>>>>>>> different algorithm. Users could pass in a custom >>>>>> `SessionWindows` as >>>>>>>>> long as the session algorithm works correctly for it. >>>>>>>>> >>>>>>>>> For the new sliding windows, we want to use a different >> algorithm >>>>>>>>> compare to hopping/tumbling windows. If we let sliding window >>>>>> extend >>>>>>>>> `Windows`, we can decide at runtime if we need to use the >>>>>>>>> hopping/tumbling window algorithm for hopping/tumbling windows >> or >>>>>> the >>>>>>>>> new sliding window algorithm for sliding windows. However, if we >>>>>> get a >>>>>>>>> custom window, which algorithm do we pick now? The existing >>>>>>>>> tumbling/hopping window algorithm of the new sliding window >>>>>> algorithm? >>>>>>>>> Both a custom "time-window" and custom "sliding window" >> implement >>>>>> the >>>>>>>>> generic `Windows` class and thus we cannot make a decision as we >>>>>> don't >>>>>>>>> know the user's intent. >>>>>>>>> >>>>>>>>> As a matter of fact, even if the user might not be aware of it, >>>>>> the >>>>>>>>> algorithm we use does already leak into the API (if a user >> extends >>>>>>>>> `Windows` is must work with our hopping/tumbling window >> algorithm >>>>>> and if >>>>>>>>> a user extends `SessionWindows` it must work with our session >>>>>> algorithm) >>>>>>>>> and it seems we need to preserve this property for sliding >> window. >>>>>>>>> >>>>>>>>> >>>>>>>>> -Matthias >>>>>>>>> >>>>>>>>> On 7/22/20 4:35 PM, Sophie Blee-Goldman wrote: >>>>>>>>>> Hey John, >>>>>>>>>> >>>>>>>>>> Just a few follow-up questions/comments about the whole >> Windows >>>>>> thing: >>>>>>>>>> >>>>>>>>>> That's a good way of looking at things; in particular the >> point >>>>>> about >>>>>>>>>> SessionWindows >>>>>>>>>> for example requiring a Merger while other "statically >>>>>> enumerable" >>>>>>>> windows >>>>>>>>>> require >>>>>>>>>> only an adder seems to touch on the heart of the matter. >>>>>>>>>> >>>>>>>>>> It seems like what Time and Universal (and any other Windows >>>>>>>>>>> implementation) have in common is that the windows are >>>>>> statically >>>>>>>>>>> enumerable. >>>>>>>>>>> As a consequence, they can all rely on an aggregation >>>>>> maintenence >>>>>>>> algorithm >>>>>>>>>>> that involves enumerating each of the windows and updating >> it. >>>>>> That >>>>>>>>>>> also means that their DSL object (TimeWindowedKStream) >> doesn't >>>>>> need >>>>>>>>>>> "subtractors" or "mergers", but only "adders"; again, this >> is a >>>>>>>> consequence >>>>>>>>>>> of the fact that the windows are enumerable. >>>>>>>>>> >>>>>>>>>> >>>>>>>>>> Given that, I'm a bit confused why you conclude that sliding >>>>>> windows >>>>>>>> are >>>>>>>>>> fundamentally >>>>>>>>>> different from the "statically enumerable" windows -- sliding >>>>>> windows >>>>>>>>>> require only an >>>>>>>>>> adder too. I'm not sure it's a consequence of being >> enumerable, >>>>>> or that >>>>>>>>>> being enumerable >>>>>>>>>> is the fundamental property that unites all Windows (ignoring >>>>>>>> JoinWindows >>>>>>>>>> here). Yes, it >>>>>>>>>> currently does apply to all Windows implementations, but we >>>>>> shouldn't >>>>>>>>>> assume that it >>>>>>>>>> *has *to be that way on the basis that it currently happens to >>>>>> be. >>>>>>>>>> >>>>>>>>>> Also, the fact that they can all rely on the same aggregation >>>>>> algorithm >>>>>>>>>> seems like an >>>>>>>>>> implementation detail and it would be weird to force a >>>>>> separate/new >>>>>>>> DSL API >>>>>>>>>> just because >>>>>>>>>> under the covers we swap in a different processor. >>>>>>>>>> >>>>>>>>>> To be fair, I don't think there's a strong reason *against* >> not >>>>>>>> extending >>>>>>>>>> Windows -- in the end >>>>>>>>>> it will just mean adding a new #windowedBy method and >>>>>> copy/pasting >>>>>>>>>> everything from >>>>>>>>>> TimeWindowedKStream pretty much word for word. But anytime >> you >>>>>> find >>>>>>>>>> yourself >>>>>>>>>> copying over code almost exactly, there should probably be a >>>>>> good >>>>>>>> reason >>>>>>>>>> why :) >>>>>>>>>> >>>>>>>>>> >>>>>>>>>> On Wed, Jul 22, 2020 at 3:48 PM John Roesler < >>>>>> vvcep...@apache.org> >>>>>>>> wrote: >>>>>>>>>> >>>>>>>>>>> Thanks Leah! >>>>>>>>>>> >>>>>>>>>>> 5) Regarding the empty windows, I'm wondering if we should >>>>>> simply >>>>>>>> propose >>>>>>>>>>> that the windows should not be emitted downstream of the >>>>>> operator or >>>>>>>>>>> visible in IQ. Then, it'll be up to the implementation to >> make >>>>>> it >>>>>>>> happen. >>>>>>>>>>> I'm >>>>>>>>>>> personally not concerned about it, since it seems like there >>>>>> are >>>>>>>> multiple >>>>>>>>>>> ways to accomplish this. >>>>>>>>>>> >>>>>>>>>>> Note, the discrepancy Matthias pointed out is actually a >>>>>> design bug. >>>>>>>> The >>>>>>>>>>> windowed aggregation (like every operation in Streams) >>>>>> produces a >>>>>>>> "view", >>>>>>>>>>> which then forms the basis of downstream operations. When we >>>>>> pass the >>>>>>>>>>> Materialized option to the operation, all we're doing is >>>>>> saying to >>>>>>>>>>> "materialize" >>>>>>>>>>> the view (aka, actually store the computed view) and also >> make >>>>>> it >>>>>>>>>>> queriable. >>>>>>>>>>> It would be illegal for the "queriable, materialized view" to >>>>>> differ >>>>>>>> in >>>>>>>>>>> any way >>>>>>>>>>> from the "view". So, it seems we must either propose to emit >>>>>> the empty >>>>>>>>>>> windows AND make them visible in IQ, or propose NOT to emit >>>>>> the empty >>>>>>>>>>> windows AND NOT make them visible in IQ. >>>>>>>>>>> >>>>>>>>>>> 7) Regarding whether we can extend TimeWindows (or Windows): >>>>>>>>>>> I've been mulling this over more. I think it's worth asking >> the >>>>>>>> question of >>>>>>>>>>> what these classes even mean. For example, why is >>>>>> SessionWindows a >>>>>>>>>>> different thing from TimeWindows and UniversalWindows (which >>>>>> are both >>>>>>>>>>> Windows)? >>>>>>>>>>> >>>>>>>>>>> This conversation is extra complicated because of the >>>>>> incomplete and >>>>>>>>>>> mis-matched class hierarchy, but we can try to look past it >>>>>> for now. >>>>>>>>>>> >>>>>>>>>>> It seems like what Time and Universal (and any other Windows >>>>>>>>>>> implementation) have in common is that the windows are >>>>>> statically >>>>>>>>>>> enumerable. >>>>>>>>>>> As a consequence, they can all rely on an aggregation >>>>>> maintenence >>>>>>>> algorithm >>>>>>>>>>> that involves enumerating each of the windows and updating >> it. >>>>>> That >>>>>>>>>>> also means that their DSL object (TimeWindowedKStream) >> doesn't >>>>>> need >>>>>>>>>>> "subtractors" or "mergers", but only "adders"; again, this >> is a >>>>>>>> consequence >>>>>>>>>>> of the fact that the windows are enumerable. >>>>>>>>>>> >>>>>>>>>>> In contrast, session windows are data driven, so they are not >>>>>>>> statically >>>>>>>>>>> enumerable. Their algorithm has to rely on scans, and to do >>>>>> the scans, >>>>>>>>>>> it needs to know the "inactivity gap", which needs to be part >>>>>> of the >>>>>>>> window >>>>>>>>>>> definition. Likewise, session windows have the property that >>>>>> they need >>>>>>>>>>> to be merged, so their DSL object also requires mergers. >>>>>>>>>>> >>>>>>>>>>> It really seems like your new window definition doesn't fit >>>>>> into >>>>>>>> either >>>>>>>>>>> category. It uses a different algorithm, which relies on >>>>>> scans, but >>>>>>>> it is >>>>>>>>>>> also fixed in size, so it doesn't need mergers. In this >>>>>> situation, it >>>>>>>> seems >>>>>>>>>>> like the safe bet is to just create SessionWindows with no >>>>>> interface >>>>>>>> and >>>>>>>>>>> add a separate set of DSL operations and objects. It's a >>>>>> little extra >>>>>>>> code, >>>>>>>>>>> but it seems to keep everything tidier and more >>>>>> comprehensible, both >>>>>>>>>>> for us and for users. >>>>>>>>>>> >>>>>>>>>>> What do you think? >>>>>>>>>>> -John >>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>>> On Wed, Jul 22, 2020, at 10:30, Leah Thomas wrote: >>>>>>>>>>>> Hi Matthias, >>>>>>>>>>>> >>>>>>>>>>>> Thanks for the suggestions, I've updated the KIP and child >>>>>> page >>>>>>>>>>> accordingly >>>>>>>>>>>> and addressed some below. >>>>>>>>>>>> >>>>>>>>>>>> 1) For the mandatory grace period, we should use a static >>>>>> builder >>>>>>>> method >>>>>>>>>>>>> that take two parameters. >>>>>>>>>>>>> >>>>>>>>>>>> >>>>>>>>>>>> That makes sense, I've changed that in the public API. >>>>>>>>>>>> >>>>>>>>>>>> Btw: this implementation actually raises an issue for IQ: >>>>>> those empty >>>>>>>>>>>>> windows would be returned. >>>>>>>>>>>> >>>>>>>>>>>> >>>>>>>>>>>> This is a great point, with the current implementation plan >>>>>> empty >>>>>>>> windows >>>>>>>>>>>> would be returned. I think creating a second window store >>>>>> would >>>>>>>>>>> definitely >>>>>>>>>>>> work, but there would be more overhead in having two stores >>>>>> and >>>>>>>> switching >>>>>>>>>>>> windows between the stores, as well as doing scans in both >>>>>> stores to >>>>>>>> find >>>>>>>>>>>> existing windows. There might be a way to do avoid emitting >>>>>> empty >>>>>>>> windows >>>>>>>>>>>> without creating a second window store, I'll look more into >>>>>> it. >>>>>>>>>>>> >>>>>>>>>>>> Cheers, >>>>>>>>>>>> Leah >>>>>>>>>>>> >>>>>>>>>>>> On Tue, Jul 21, 2020 at 1:25 PM Matthias J. Sax < >>>>>> mj...@apache.org> >>>>>>>>>>> wrote: >>>>>>>>>>>> >>>>>>>>>>>>> Thanks for updating the KIP. >>>>>>>>>>>>> >>>>>>>>>>>>> Couple of follow up comments: >>>>>>>>>>>>> >>>>>>>>>>>>> 1) For the mandatory grace period, we should use a static >>>>>> builder >>>>>>>>>>> method >>>>>>>>>>>>> that take two parameters. This provides a better API as >> user >>>>>> cannot >>>>>>>>>>>>> forget to set the grace period. Throwing a runtime >> exception >>>>>> seems >>>>>>>> not >>>>>>>>>>>>> to be the best way to handle this case. >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> 2) In Fig.2 you list 10 hopping windows. I believe it >> should >>>>>>>> actually >>>>>>>>>>> be >>>>>>>>>>>>> more? There first hopping window would be [-6,-4[ and the >>>>>> last one >>>>>>>>>>> would >>>>>>>>>>>>> be from [19,29[ -- hence, the cost saving are actually much >>>>>> higher. >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> 3a) IQ: you are saying that the user need to compute the >>>>>> start time >>>>>>>> as >>>>>>>>>>>>> >>>>>>>>>>>>>> windowSize+the time they're looking at >>>>>>>>>>>>> >>>>>>>>>>>>> Should this be "targetTime - windowSize" instead? >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> 3b) IQ: in you example you say "window size of 10 minutes" >>>>>> with an >>>>>>>>>>>>> incident at 9:15. >>>>>>>>>>>>> >>>>>>>>>>>>>> they're looking for a window with the start time of 8:15. >>>>>>>>>>>>> >>>>>>>>>>>>> The example does not seem to add up? >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> 4) For "Processing Windows": you describe a three step >>>>>> approach: I >>>>>>>> just >>>>>>>>>>>>> want to point out, that step (1) is not necessary for each >>>>>> input >>>>>>>>>>> record, >>>>>>>>>>>>> because timestamps are not guaranteed to be unique and >> thus a >>>>>>>> previous >>>>>>>>>>>>> record with the same key and timestamp might have create >> the >>>>>> windows >>>>>>>>>>>>> already. >>>>>>>>>>>>> >>>>>>>>>>>>> Nit: I am also not exactly sure what you mean by step (3) >> as >>>>>> you use >>>>>>>>>>> the >>>>>>>>>>>>> word "send". I guess you mean "put"? >>>>>>>>>>>>> >>>>>>>>>>>>> It seem there are actually more details in the sub-page: >>>>>>>>>>>>> >>>>>>>>>>>>>> A new record for SlidingWindows will always create two new >>>>>> windows. >>>>>>>>>>> If >>>>>>>>>>>>> either of those windows already exist in the windows store, >>>>>> their >>>>>>>>>>>>> aggregation will simply be updated to include the new >>>>>> record, but no >>>>>>>>>>>>> duplicate window will be added to the WindowStore. >>>>>>>>>>>>> >>>>>>>>>>>>> However, the first and second sentence contradict each >> other >>>>>> a >>>>>>>> little >>>>>>>>>>>>> bit. I think the first sentence is not correct. >>>>>>>>>>>>> >>>>>>>>>>>>> Nit: >>>>>>>>>>>>> >>>>>>>>>>>>>> For in-order records, the left window will always be >> empty. >>>>>>>>>>>>> >>>>>>>>>>>>> This should be "right window" ? >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> 5) "Emitting Results": it might be worth to point out, >> that a >>>>>>>>>>>>> second/future window of a new record is create with no >>>>>> records, and >>>>>>>>>>>>> thus, even if it's initialized it won't be emitted. Only >> if a >>>>>>>>>>>>> consecutive record falls into the window, the window would >> be >>>>>>>> updates >>>>>>>>>>>>> and the window result (for a window content of one record) >>>>>> would be >>>>>>>>>>> sent >>>>>>>>>>>>> downstream. >>>>>>>>>>>>> >>>>>>>>>>>>> Again, the sub-page contains this details. Might still be >>>>>> worth to >>>>>>>> add >>>>>>>>>>>>> to the top level page, too. >>>>>>>>>>>>> >>>>>>>>>>>>> Btw: this implementation actually raises an issue for IQ: >>>>>> those >>>>>>>> empty >>>>>>>>>>>>> windows would be returned. Thus I am wondering if we need >> to >>>>>> use two >>>>>>>>>>>>> stores internally? One store for actual windows and one >>>>>> store for >>>>>>>> empty >>>>>>>>>>>>> windows? If an empty window is updated, it's move to the >>>>>> other >>>>>>>> store? >>>>>>>>>>>>> For IQ, we only allow to query the non-empty-window store? >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> 6) On the sub-page: >>>>>>>>>>>>> >>>>>>>>>>>>>> The left window of in-order records and both windows for >>>>>>>> out-of-order >>>>>>>>>>>>> records need to be updated with the values of records that >>>>>> have >>>>>>>> already >>>>>>>>>>>>> been processed. >>>>>>>>>>>>> >>>>>>>>>>>>> Why "both windows for out-of-order records"? IMHO, we don't >>>>>> know how >>>>>>>>>>>>> many existing windows needs to be updated when processing >> an >>>>>>>>>>>>> out-of-order record. Of course, an out-of-order record >> could >>>>>> not >>>>>>>> fall >>>>>>>>>>>>> into any existing window but create two new windows, too. >>>>>>>>>>>>> >>>>>>>>>>>>>> Because each record creates one new window that includes >>>>>> itself >>>>>>>> and >>>>>>>>>>> one >>>>>>>>>>>>> window that does not >>>>>>>>>>>>> >>>>>>>>>>>>> As state above, this does not seem to hold. I understand >> why >>>>>> you >>>>>>>> mean, >>>>>>>>>>>>> but it would be good to be exact. >>>>>>>>>>>>> >>>>>>>>>>>>> Figure 2: You use the term "late" but you mean >>>>>> "out-of-order" I >>>>>>>> guess >>>>>>>>>>> -- >>>>>>>>>>>>> a record is _late_ if it's not processed any longer as the >>>>>> grace >>>>>>>> period >>>>>>>>>>>>> passed already. >>>>>>>>>>>>> >>>>>>>>>>>>> Figure 2: "Late" should be out-or-order. The example text >>>>>> say a >>>>>>>> window >>>>>>>>>>>>> [16,26] should be created but the figure shows the green >>>>>> window as >>>>>>>>>>> [15,20]. >>>>>>>>>>>>> >>>>>>>>>>>>> About the blue window: maybe add not that the blue window >>>>>> contains >>>>>>>> the >>>>>>>>>>>>> aggregate we need for the green window, _before_ the new >>>>>> record `a` >>>>>>>> is >>>>>>>>>>>>> added to the blue window. >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> 7) I am not really happy to extend TimeWindows and I think >>>>>> the >>>>>>>> argument >>>>>>>>>>>>> about JoinWindows is not the best (IMHO, JoinWindows do it >>>>>> already >>>>>>>>>>> wrong >>>>>>>>>>>>> and we just repeat the same mistake). However, it seems our >>>>>> window >>>>>>>>>>>>> hierarchy is "broken" already and it might be out of scope >>>>>> for this >>>>>>>> KIP >>>>>>>>>>>>> to fix it. Hence, I am ok that we bite the bullet for now >>>>>> and clean >>>>>>>> it >>>>>>>>>>>>> up later. >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> -Matthias >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> On 7/20/20 5:18 PM, Guozhang Wang wrote: >>>>>>>>>>>>>> Hi Leah, >>>>>>>>>>>>>> >>>>>>>>>>>>>> Thanks for the updated KIP. I agree that extending >>>>>> SlidingWindows >>>>>>>>>>> from >>>>>>>>>>>>>> Windows is fine for the sake of not introducing more >> public >>>>>> APIs >>>>>>>> (and >>>>>>>>>>>>> their >>>>>>>>>>>>>> internal xxxImpl classes), and its cons is small enough to >>>>>> tolerate >>>>>>>>>>> to >>>>>>>>>>>>> me. >>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>> Guozhang >>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>> On Mon, Jul 20, 2020 at 1:49 PM Leah Thomas < >>>>>> ltho...@confluent.io> >>>>>>>>>>>>> wrote: >>>>>>>>>>>>>> >>>>>>>>>>>>>>> Hi all, >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> Thanks for the feedback on the KIP. I've updated the KIP >>>>>> page >>>>>>>>>>>>>>> < >>>>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>> >>>>>>>> >>>>>> >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-450%3A+Sliding+Window+Aggregations+in+the+DSL >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>> to address these points and have created a child page >>>>>>>>>>>>>>> < >>>>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>> >>>>>>>> >>>>>> >> https://cwiki.apache.org/confluence/display/KAFKA/Aggregation+details+for+KIP-450 >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>> to go more in depth on certain implementation details. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> *Grace Period:* >>>>>>>>>>>>>>> I think Sophie raises a good point that the default grace >>>>>> period >>>>>>>> of >>>>>>>>>>> 24 >>>>>>>>>>>>>>> hours is often too long and was chosen when retention >> time >>>>>> and >>>>>>>> grace >>>>>>>>>>>>> period >>>>>>>>>>>>>>> were the same. For SlidingWindows, I propose we make the >>>>>> grace >>>>>>>>>>> period >>>>>>>>>>>>>>> mandatory. To keep formatting consistent with other types >>>>>> of >>>>>>>>>>> windows, >>>>>>>>>>>>> grace >>>>>>>>>>>>>>> period won't be an additional parameter in the #of >> method, >>>>>> but >>>>>>>> will >>>>>>>>>>>>> still >>>>>>>>>>>>>>> look like it does in other use cases: >>>>>>>>>>>>>>> >>>>>> .windowedBy(SlidingWindows.of(twentySeconds).grace(fiftySeconds). >>>>>>>> If >>>>>>>>>>>>> grace >>>>>>>>>>>>>>> period isn't properly initialized, an error will be >> thrown >>>>>> through >>>>>>>>>>> the >>>>>>>>>>>>>>> process method. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> *Storage Layer + Aggregation:* >>>>>>>>>>>>>>> SlidingWindows will use a WindowStore because computation >>>>>> can be >>>>>>>>>>> done >>>>>>>>>>>>> with >>>>>>>>>>>>>>> the information stored in a WindowStore (window timestamp >>>>>> and >>>>>>>>>>> value). >>>>>>>>>>>>> Using >>>>>>>>>>>>>>> the WindowStore also simplifies computation as >>>>>> SlidingWindows can >>>>>>>>>>>>> leverage >>>>>>>>>>>>>>> existing processes. Because we are using a WindowStore, >> the >>>>>>>>>>> aggregation >>>>>>>>>>>>>>> process will be similar to that of a hopping window. As >>>>>> records >>>>>>>>>>> come in >>>>>>>>>>>>>>> their value is added to the aggregation that already >>>>>> exists, >>>>>>>>>>> following >>>>>>>>>>>>> the >>>>>>>>>>>>>>> same procedure as hopping windows. The aggregation >>>>>> difference >>>>>>>>>>> between >>>>>>>>>>>>>>> SlidingWindows and HoppingWindows comes in creating new >>>>>> windows >>>>>>>> for >>>>>>>>>>> a >>>>>>>>>>>>>>> SlidingWindow, where you need to find the existing >> records >>>>>> that >>>>>>>>>>> belong >>>>>>>>>>>>> to >>>>>>>>>>>>>>> the new window. This computation is similar to the >>>>>> aggregation in >>>>>>>>>>>>>>> SessionWindows and requires a scan to the WindowStore to >>>>>> find the >>>>>>>>>>> window >>>>>>>>>>>>>>> with the aggregation needed, which will always be >>>>>> pre-computed. >>>>>>>> The >>>>>>>>>>> scan >>>>>>>>>>>>>>> requires creating an iterator, but should have minimal >>>>>> performance >>>>>>>>>>>>> effects >>>>>>>>>>>>>>> as this strategy is already implemented in >> SessionWindows. >>>>>> More >>>>>>>>>>> details >>>>>>>>>>>>> on >>>>>>>>>>>>>>> finding the aggregation that needs to be put in a new >>>>>> window can >>>>>>>> be >>>>>>>>>>>>> found >>>>>>>>>>>>>>> on the implementation page >>>>>>>>>>>>>>> < >>>>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>> >>>>>>>> >>>>>> >> https://cwiki.apache.org/confluence/display/KAFKA/Aggregation+details+for+KIP-450 >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>> . >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> *Extending Windows<TimeWindow>, Windows<Window> or >> nothing* >>>>>>>>>>>>>>> Because SlidingWindows are still defined by a windowSize >>>>>> (whereas >>>>>>>>>>>>>>> SessionWindows are defined purely by data), I think it >>>>>> makes sense >>>>>>>>>>> to >>>>>>>>>>>>>>> leverage the existing Window processes instead of >> creating >>>>>> a new >>>>>>>>>>> store >>>>>>>>>>>>> type >>>>>>>>>>>>>>> that would be very similar to the WindowStore. While it's >>>>>> true >>>>>>>> that >>>>>>>>>>> the >>>>>>>>>>>>>>> #windowsFor method isn't necessary for SlidingWindows, >>>>>> JoinWindows >>>>>>>>>>> also >>>>>>>>>>>>>>> extends Windows<Window> and throws an >>>>>>>> UnsupportedOperationException >>>>>>>>>>> in >>>>>>>>>>>>> the >>>>>>>>>>>>>>> #windowsFor method, which is what SlidingWindows can do. >>>>>> The >>>>>>>>>>> difference >>>>>>>>>>>>>>> between extending Windows<TimeWindow> or Windows<Window> >> is >>>>>>>>>>> minimal, as >>>>>>>>>>>>>>> both are ways to pass window parameters. Extending >>>>>>>>>>> Windows<TimeWindow> >>>>>>>>>>>>> will >>>>>>>>>>>>>>> give us more leverage in utilizing existing processes. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> *Emit Strategy* >>>>>>>>>>>>>>> I would argue that emitting for every update is still the >>>>>> best way >>>>>>>>>>> to go >>>>>>>>>>>>>>> for SlidingWindows because it mimics the other types of >>>>>> windows, >>>>>>>> and >>>>>>>>>>>>>>> suppression can be leveraged to limit what SlidingWindows >>>>>> emits. >>>>>>>>>>> While >>>>>>>>>>>>> some >>>>>>>>>>>>>>> users may only want to see the last value, others may >> want >>>>>> to see >>>>>>>>>>> more, >>>>>>>>>>>>> and >>>>>>>>>>>>>>> leaving the emit strategy to emit partial results allows >>>>>> both >>>>>>>> users >>>>>>>>>>> to >>>>>>>>>>>>>>> access what they want. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> *Additional Features* >>>>>>>>>>>>>>> Supporting sliding windows inherently, and shifting >>>>>> inefficient >>>>>>>>>>> hopping >>>>>>>>>>>>>>> windows to sliding windows, is an interesting idea and >>>>>> could be >>>>>>>>>>> built on >>>>>>>>>>>>>>> top of SlidingWindows when they are finished, but right >>>>>> now seems >>>>>>>>>>> out of >>>>>>>>>>>>>>> scope for the needs of this KIP. Similarly, including a >>>>>>>>>>> `subtraction` >>>>>>>>>>>>>>> feature could have performance improvements, but doesn't >>>>>> seem >>>>>>>>>>> necessary >>>>>>>>>>>>> for >>>>>>>>>>>>>>> the implementation of this KIP. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> Let me know what you think of the updates, >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> Leah >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> On Thu, Jul 16, 2020 at 11:57 AM John Roesler < >>>>>>>> vvcep...@apache.org> >>>>>>>>>>>>> wrote: >>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> Hello all, >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> Thanks for the KIP, Leah! >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> Regarding (1): I'd go farther actually. Making Windows >> an >>>>>>>> abstract >>>>>>>>>>>>>>>> class was a mistake from the beginning that led to us >> not >>>>>> being >>>>>>>>>>>>>>>> able to fix a very confusing situation for users around >>>>>> retention >>>>>>>>>>>>> times, >>>>>>>>>>>>>>>> final results emitting, etc. Thus, I would not suggest >>>>>> extending >>>>>>>>>>>>>>>> TimeWindows for sure, but would also not suggest >> extending >>>>>>>> Windows. >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> The very simplest thing to do is follow the example of >>>>>>>>>>> SessionWindows, >>>>>>>>>>>>>>>> which is just a completely self-contained class. If we >>>>>> don't mess >>>>>>>>>>> with >>>>>>>>>>>>>>>> class inheritance, we won't ever have any of the >> problems >>>>>> related >>>>>>>>>>> to >>>>>>>>>>>>>>>> class inheritance. This is my preferred solution. >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> Still, Sliding windows has a lot in common with >>>>>> TimeWindows and >>>>>>>>>>> other >>>>>>>>>>>>>>>> fixed-size windows, namely that the windows are fixed in >>>>>> size. If >>>>>>>>>>> we >>>>>>>>>>>>> want >>>>>>>>>>>>>>>> to preserve the current two-part windowing API in which >>>>>> you can >>>>>>>>>>> window >>>>>>>>>>>>>>>> by either "fixed" or "data driven" modes, I'd suggest we >>>>>> avoid >>>>>>>>>>>>> increasing >>>>>>>>>>>>>>>> the blast radius of Windows by taking the opportunity to >>>>>> replace >>>>>>>> it >>>>>>>>>>>>> with >>>>>>>>>>>>>>>> a proper interface and implement that interface instead. >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> For example: >>>>>>>>>>>>>>>> https://github.com/apache/kafka/pull/9031 >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> Then, SlidingWindows would just implement >>>>>>>> FixedSizeWindowDefinition >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> ====== >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> Regarding (2), it seems more straightforward as a user >> of >>>>>> Streams >>>>>>>>>>>>>>>> to just have one mental model. _All_ of our aggregation >>>>>>>> operations >>>>>>>>>>>>>>>> follow an eager emission model, in which we just emit an >>>>>> update >>>>>>>>>>>>> whenever >>>>>>>>>>>>>>>> an update is available. We already provided Suppression >> to >>>>>>>>>>> explicitly >>>>>>>>>>>>>>> apply >>>>>>>>>>>>>>>> different update semantics in the case it's required. >> Why >>>>>> should >>>>>>>> we >>>>>>>>>>>>>>> define >>>>>>>>>>>>>>>> a snowflake operation with completely different >> semantics >>>>>> from >>>>>>>>>>>>> everything >>>>>>>>>>>>>>>> else? I.e., systems are generally easier to use when >> they >>>>>> follow >>>>>>>> a >>>>>>>>>>> few >>>>>>>>>>>>>>>> simple, composable rules than when they have a lot of >>>>>> different, >>>>>>>>>>>>> specific >>>>>>>>>>>>>>>> rules. >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> ====== >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> New point: (4): >>>>>>>>>>>>>>>> It would be nice to include some examples of user code >>>>>> that would >>>>>>>>>>> use >>>>>>>>>>>>> the >>>>>>>>>>>>>>>> new API, which should include: >>>>>>>>>>>>>>>> 1. using the DSL with the sliding window definition >>>>>>>>>>>>>>>> 2. accessing the stored results of a sliding window >>>>>> aggregation >>>>>>>>>>> via IQ >>>>>>>>>>>>>>>> 3. defining a custom processor to access sliding windows >>>>>> in a >>>>>>>> store >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> It generally helps reviewers wrap their heads around the >>>>>>>> proposal, >>>>>>>>>>> as >>>>>>>>>>>>>>> well >>>>>>>>>>>>>>>> as shaking out any design issues that would otherwise >>>>>> only come >>>>>>>> up >>>>>>>>>>>>> during >>>>>>>>>>>>>>>> implementation/testing/review. >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> Thanks again for the awesome proposal! >>>>>>>>>>>>>>>> -John >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> On Tue, Jul 14, 2020, at 12:31, Guozhang Wang wrote: >>>>>>>>>>>>>>>>> Hello Leah, >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> Thanks for the nice written KIP. A few thoughts: >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> 1) I echo the other reviewer's comments regarding the >>>>>> typing: >>>>>>>> why >>>>>>>>>>>>>>>> extending >>>>>>>>>>>>>>>>> TimeWindow instead of just extending Window? >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> 2) I also feel that emitting policy for this type of >>>>>> windowing >>>>>>>>>>>>>>>> aggregation >>>>>>>>>>>>>>>>> may be different from the existing ones. Existing >>>>>> emitting >>>>>>>> policy >>>>>>>>>>> is >>>>>>>>>>>>>>> very >>>>>>>>>>>>>>>>> simple: emit every time when window get updates, and >>>>>> emit every >>>>>>>>>>> time >>>>>>>>>>>>> on >>>>>>>>>>>>>>>>> out-of-ordering data within grace period, this is >>>>>> because for >>>>>>>>>>>>>>>> time-windows >>>>>>>>>>>>>>>>> the window close time is strictly depend on the window >>>>>> start >>>>>>>> time >>>>>>>>>>>>> which >>>>>>>>>>>>>>>> is >>>>>>>>>>>>>>>>> fixed, while for session-windows although the window >>>>>> open/close >>>>>>>>>>> time >>>>>>>>>>>>> is >>>>>>>>>>>>>>>>> also data-dependent it is relatively infrequent >> compared >>>>>> to the >>>>>>>>>>>>>>>>> sliding-windows. For this KIP, since each new data >> would >>>>>> cause a >>>>>>>>>>>>>>>>> new sliding-window, the num. windows maintained >>>>>> logically could >>>>>>>> be >>>>>>>>>>>>> much >>>>>>>>>>>>>>>>> larger and hence emitting on each update may be too >>>>>> aggressive. >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> 3) Although KIP itself should be focusing on user face >>>>>>>>>>> interfaces, I'd >>>>>>>>>>>>>>>>> suggest we create a children page of KIP-450 discussing >>>>>> about >>>>>>>> its >>>>>>>>>>>>>>>>> implementations as well, since some of that may drive >> the >>>>>>>>>>> interface >>>>>>>>>>>>>>>> design. >>>>>>>>>>>>>>>>> E.g. personally I think having a combiner interface in >>>>>> addition >>>>>>>> to >>>>>>>>>>>>>>>>> aggregator would be useful but that's based on my >> 2cents >>>>>> about >>>>>>>> the >>>>>>>>>>>>>>>>> implementation design (I once created a child page >>>>>> describing >>>>>>>> it: >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>> >>>>>>>> >>>>>> >> https://cwiki.apache.org/confluence/display/KAFKA/Optimize+Windowed+Aggregation >>>>>>>>>>>>>>>>> ). >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> Guozhang >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> On Tue, Jul 14, 2020 at 5:31 AM Bruno Cadonna < >>>>>>>> br...@confluent.io >>>>>>>>>>>> >>>>>>>>>>>>>>>> wrote: >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> Hi Leah, >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> Thank you for the KIP! >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> Here is my feedback: >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> 1. The KIP would benefit from some code examples that >>>>>> show how >>>>>>>>>>> to use >>>>>>>>>>>>>>>>>> sliding windows in aggregations. >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> 2. The different sliding windows in Figure 1 and 2 are >>>>>> really >>>>>>>>>>> hard to >>>>>>>>>>>>>>>>>> distinguish. Could you please try to make them >>>>>> graphically >>>>>>>> better >>>>>>>>>>>>>>>>>> distinguishable? You could try to draw the frames of >>>>>>>> consecutive >>>>>>>>>>>>>>>>>> windows shifted to each other. >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> 3. I agree with Matthias, that extending >>>>>> Windows<TimeWindow> >>>>>>>>>>> does not >>>>>>>>>>>>>>>>>> seem to be the best approach. What would be the result >>>>>> of >>>>>>>>>>>>>>>>>> windowsFor()? >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> 4. In the section "Public Interfaces" you should >> remove >>>>>>>>>>>>>>> implementation >>>>>>>>>>>>>>>>>> details like private constructors and private fields. >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> 5. Do we need a new store interface or can we use >>>>>> WindowStore? >>>>>>>>>>> Some >>>>>>>>>>>>>>>>>> words about that would be informative. >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> 6. @Matthias, if the subtrator is not strictly needed, >>>>>> I would >>>>>>>>>>> skip >>>>>>>>>>>>>>> it >>>>>>>>>>>>>>>>>> for now and add it later. >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> 7. I also agree that having a section that describes >>>>>> how to >>>>>>>>>>> handle >>>>>>>>>>>>>>>>>> out-of-order records would be good to understand what >>>>>> is still >>>>>>>>>>>>>>> missing >>>>>>>>>>>>>>>>>> and what we can reuse. >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> Best, >>>>>>>>>>>>>>>>>> Bruno >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> On Sat, Jul 11, 2020 at 9:16 PM Matthias J. Sax < >>>>>>>>>>> mj...@apache.org> >>>>>>>>>>>>>>>> wrote: >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> Leah, >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> thanks for your update. However, it does not >>>>>> completely answer >>>>>>>>>>> my >>>>>>>>>>>>>>>>>> question. >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> In our current window implementations, we emit a >>>>>> window result >>>>>>>>>>>>>>> update >>>>>>>>>>>>>>>>>>> record (ie, early/partial result) for each input >>>>>> record. When >>>>>>>> an >>>>>>>>>>>>>>>>>>> out-of-order record arrives, we just update to >>>>>> corresponding >>>>>>>> old >>>>>>>>>>>>>>>> window >>>>>>>>>>>>>>>>>>> and emit another update. >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> It's unclear from the KIP if you propose the same >> emit >>>>>>>>>>> strategy? -- >>>>>>>>>>>>>>>> For >>>>>>>>>>>>>>>>>>> sliding windows it might be worth to consider to use >> a >>>>>>>> different >>>>>>>>>>>>>>> emit >>>>>>>>>>>>>>>>>>> strategy and only support emitting the final result >>>>>> only (ie, >>>>>>>>>>> after >>>>>>>>>>>>>>>> the >>>>>>>>>>>>>>>>>>> grace period passed)? >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> Boyang, also raises a good point that relates to my >>>>>> point from >>>>>>>>>>>>>>> above >>>>>>>>>>>>>>>>>>> about pre-aggregations and storage layout. Our >> current >>>>>> time >>>>>>>>>>> windows >>>>>>>>>>>>>>>> are >>>>>>>>>>>>>>>>>>> all pre-aggregated and stored in parallel. We can >> also >>>>>> lookup >>>>>>>>>>>>>>> windows >>>>>>>>>>>>>>>>>>> efficiently, as we can compute the windowed-key given >>>>>> the >>>>>>>> input >>>>>>>>>>>>>>>> record >>>>>>>>>>>>>>>>>>> key and timestamp based on the window definition. >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> However, for sliding windows, window boundaries are >>>>>> data >>>>>>>>>>> dependent >>>>>>>>>>>>>>>> and >>>>>>>>>>>>>>>>>>> thus we cannot compute them upfront. Thus, how can we >>>>>> "find" >>>>>>>>>>>>>>> existing >>>>>>>>>>>>>>>>>>> window efficiently? Furthermore, out-of-order data >>>>>> would >>>>>>>> create >>>>>>>>>>> new >>>>>>>>>>>>>>>>>>> windows in the past and we need to be able to handle >>>>>> this >>>>>>>> case. >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> Thus, to handle out-of-order data correctly, we need >>>>>> to store >>>>>>>>>>> all >>>>>>>>>>>>>>> raw >>>>>>>>>>>>>>>>>>> input events. Additionally, we could also store >>>>>> pre-aggregated >>>>>>>>>>>>>>>> results >>>>>>>>>>>>>>>>>>> if we thinks it's benfitial. -- If we apply "emit >> only >>>>>> final >>>>>>>>>>>>>>> results" >>>>>>>>>>>>>>>>>>> strategy, storing pre-aggregated result would not be >>>>>> necessary >>>>>>>>>>>>>>>> though. >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> Btw: for sliding windows it might also be useful to >>>>>> consider >>>>>>>>>>>>>>> allowing >>>>>>>>>>>>>>>>>>> users to supply a `Subtractor` -- this subtractor >>>>>> could be >>>>>>>>>>> applied >>>>>>>>>>>>>>> on >>>>>>>>>>>>>>>>>>> the current window result (in case we store it) if a >>>>>> record >>>>>>>>>>> drops >>>>>>>>>>>>>>>> out of >>>>>>>>>>>>>>>>>>> the window. Of course, not all aggregation functions >>>>>> are >>>>>>>>>>>>>>> subtractable >>>>>>>>>>>>>>>>>>> and we can consider this as a follow up task, too, >> and >>>>>> not >>>>>>>>>>> include >>>>>>>>>>>>>>> in >>>>>>>>>>>>>>>>>>> this KIP for now. Thoughts? >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> I was also thinking about the type hierarchy. I am >> not >>>>>> sure if >>>>>>>>>>>>>>>> extending >>>>>>>>>>>>>>>>>>> TimeWindow is the best approach? For TimeWindows, we >>>>>> can >>>>>>>>>>>>>>> pre-compute >>>>>>>>>>>>>>>>>>> window boundaries (cf `windowsFor()`) while for a >>>>>> sliding >>>>>>>> window >>>>>>>>>>>>>>> the >>>>>>>>>>>>>>>>>>> boundaries are data dependent. Session windows are >>>>>> also data >>>>>>>>>>>>>>>> dependent >>>>>>>>>>>>>>>>>>> and thus they don't inherit from TimeWindow (Maybe >>>>>> check out >>>>>>>> the >>>>>>>>>>>>>>> KIP >>>>>>>>>>>>>>>>>>> that added session windows? It could provides some >> good >>>>>>>>>>> insights.) >>>>>>>>>>>>>>>> -- I >>>>>>>>>>>>>>>>>>> believe the same rational applies to sliding windows? >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> -Matthias >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> On 7/10/20 12:47 PM, Boyang Chen wrote: >>>>>>>>>>>>>>>>>>>> Thanks Leah and Sophie for the KIP. >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> 1. I'm a bit surprised that we don't have an advance >>>>>> time. >>>>>>>>>>> Could >>>>>>>>>>>>>>> we >>>>>>>>>>>>>>>>>>>> elaborate how the storage layer is structured? >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> 2. IIUC, there will be extra cost in terms of >> fetching >>>>>>>>>>>>>>> aggregation >>>>>>>>>>>>>>>>>> results, >>>>>>>>>>>>>>>>>>>> since we couldn't pre-aggregate until the user asks >>>>>> for it. >>>>>>>>>>> Would >>>>>>>>>>>>>>>> be >>>>>>>>>>>>>>>>>> good >>>>>>>>>>>>>>>>>>>> to also discuss it. >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> 3. We haven't discussed the possibility of >> supporting >>>>>> sliding >>>>>>>>>>>>>>>> windows >>>>>>>>>>>>>>>>>>>> inherently. For a user who actually uses a hopping >>>>>> window, >>>>>>>>>>>>>>> Streams >>>>>>>>>>>>>>>>>> could >>>>>>>>>>>>>>>>>>>> detect such an inefficiency doing a >>>>>> window_size/advance_time >>>>>>>>>>>>>>> ratio >>>>>>>>>>>>>>>> to >>>>>>>>>>>>>>>>>> reach >>>>>>>>>>>>>>>>>>>> a conclusion on whether the write amplification is >>>>>> too high >>>>>>>>>>>>>>>> compared >>>>>>>>>>>>>>>>>> with >>>>>>>>>>>>>>>>>>>> some configured threshold. The benefit of doing so >> is >>>>>> that >>>>>>>>>>>>>>> existing >>>>>>>>>>>>>>>>>> Streams >>>>>>>>>>>>>>>>>>>> users don't need to change their code, learn a new >>>>>> API, but >>>>>>>>>>> only >>>>>>>>>>>>>>> to >>>>>>>>>>>>>>>>>> upgrade >>>>>>>>>>>>>>>>>>>> Streams library to get benefits for their >> inefficient >>>>>> hopping >>>>>>>>>>>>>>>> window >>>>>>>>>>>>>>>>>>>> implementation. There might be some compatibility >>>>>> issues for >>>>>>>>>>>>>>> sure, >>>>>>>>>>>>>>>> but >>>>>>>>>>>>>>>>>>>> worth listing them out for trade-off. >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> Boyang >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> On Fri, Jul 10, 2020 at 12:40 PM Leah Thomas < >>>>>>>>>>>>>>> ltho...@confluent.io >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> wrote: >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>> Hey Matthias, >>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>> Thanks for pointing that out. I added the following >>>>>> to the >>>>>>>>>>>>>>> Propose >>>>>>>>>>>>>>>>>> Changes >>>>>>>>>>>>>>>>>>>>> section of the KIP: >>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>> "Records that come out of order will be processed >>>>>> the same >>>>>>>> way >>>>>>>>>>>>>>> as >>>>>>>>>>>>>>>>>> in-order >>>>>>>>>>>>>>>>>>>>> records, as long as they fall within the grace >>>>>> period. Any >>>>>>>> new >>>>>>>>>>>>>>>> windows >>>>>>>>>>>>>>>>>>>>> created by the late record will still be created, >>>>>> and the >>>>>>>>>>>>>>> existing >>>>>>>>>>>>>>>>>> windows >>>>>>>>>>>>>>>>>>>>> that are changed by the late record will be >> updated. >>>>>> Any >>>>>>>>>>> record >>>>>>>>>>>>>>>> that >>>>>>>>>>>>>>>>>> falls >>>>>>>>>>>>>>>>>>>>> outside of the grace period (either user defined or >>>>>> default) >>>>>>>>>>>>>>> will >>>>>>>>>>>>>>>> be >>>>>>>>>>>>>>>>>>>>> discarded. " >>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>> All the best, >>>>>>>>>>>>>>>>>>>>> Leah >>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>> On Thu, Jul 9, 2020 at 9:47 PM Matthias J. Sax < >>>>>>>>>>>>>>> mj...@apache.org> >>>>>>>>>>>>>>>>>> wrote: >>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>> Leah, >>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>> thanks a lot for the KIP. Very well written. >>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>> The KIP does not talk about the handling of >>>>>> out-of-order >>>>>>>> data >>>>>>>>>>>>>>>> though. >>>>>>>>>>>>>>>>>>>>>> How do you propose to address this? >>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>> -Matthias >>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>> On 7/8/20 5:33 PM, Leah Thomas wrote: >>>>>>>>>>>>>>>>>>>>>>> Hi all, >>>>>>>>>>>>>>>>>>>>>>> I'd like to kick-off the discussion for KIP-450, >>>>>> adding >>>>>>>>>>>>>>> sliding >>>>>>>>>>>>>>>>>> window >>>>>>>>>>>>>>>>>>>>>>> aggregation support to Kafka Streams. >>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>> >>>>>>>> >>>>>> >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-450%3A+Sliding+Window+Aggregations+in+the+DSL >>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>> Let me know what you think, >>>>>>>>>>>>>>>>>>>>>>> Leah >>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> -- >>>>>>>>>>>>>>>>> -- Guozhang >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> Attachments: >>>>>>>>> * signature.asc >>>>>>>> >>>>>>> >>>>>> >>>>> >> >
signature.asc
Description: OpenPGP digital signature
