Sorry for the late reply Matthias. Have been busy with other work recently. I'll restart the discussion and update the KIP accordingly.
Lei On Tue, Nov 6, 2018 at 3:11 PM Matthias J. Sax <matth...@confluent.io> wrote: > Any update on this KIP? > > On 9/20/18 3:30 PM, Matthias J. Sax wrote: > > Thanks for following up. Very nice examples! > > > > I think, that the window definition for Flink is semantically > > questionable. If there is only a single record, why is the window > > defined as [ts, ts+gap]? To me, this definition is not sound and seems > > to be arbitrary. To define the windows as [ts-gap,ts+gap] as you mention > > would be semantically more useful -- still, I think that defining the > > window as [ts,ts] as we do currently in Kafka Streams is semantically > > the best. > > > > I have the impression, that Flink only defines them differently, because > > it solves the issues in the implementation. (Ie, an implementation > > details leaks into the semantics, what is usually not desired.) > > > > However, I believe that we could change the implementation accordingly. > > We could store the windowed keys, as [ts-gap,ts+gap] (or [ts,ts+gap]) in > > RocksDB, but at API level we return [ts,ts]. This way, we can still find > > all windows we need and provide the same deterministic behavior and keep > > the current window boundaries on the semantic level (there is no need to > > store the window start and/or end time). With this technique, we can > > also implement dynamic session gaps. I think, we would need to store the > > used "gap" for each window, too. But again, this would be an > > implementation detail. > > > > Let's see what others think. > > > > One tricky question we would need to address is, how we can be backward > > compatible. I am currently working on KIP-258 that should help to > > address this backward compatibility issue though. > > > > > > -Matthias > > > > > > > > On 9/19/18 5:17 PM, Lei Chen wrote: > >> Thanks Matthias. That makes sense. > >> > >> You're right that symmetric merge is necessary to ensure consistency. On > >> the other hand, I kinda feel it defeats the purpose of dynamic gap, > which > >> is to update the gap from old value to new value. The symmetric merge > >> always honor the larger gap in both direction, rather than honor the gap > >> carried by record with larger timestamp. I wasn't able to find any > semantic > >> definitions w.r.t this particular aspect online, but spent some time > >> looking into other streaming engines like Apache Flink. > >> > >> Apache Flink defines the window differently, that uses (start time, > start > >> time + gap). > >> > >> so our previous example (10, 10), (19,5),(15,3) in Flink's case will be: > >> [10,20] > >> [19,24] => merged to [10,24] > >> [15,18] => merged to [10,24] > >> > >> while example (15,3)(19,5)(10,10) will be > >> [15,18] > >> [19,24] => no merge > >> [10,20] => merged to [10,24] > >> > >> however, since it only records gap in future direction, not past, a late > >> record might not trigger any merge where in symmetric merge it would. > >> (7,2),(10, 10), (19,5),(15,3) > >> [7,9] > >> [10,20] > >> [19,24] => merged to [10,24] > >> [15,18] => merged to [10,24] > >> so at the end > >> two windows [7,9][10,24] are there. > >> > >> As you can see, in Flink, the gap semantic is more toward to the way > that, > >> a gap carried by one record only affects how this record merges with > future > >> records. e.g. a later event (T2, G2) will only be merged with (T1, G1) > is > >> T2 is less than T1+G1, but not when T1 is less than T2 - G2. Let's call > >> this "forward-merge" way of handling this. I just went thought some > source > >> code and if my understanding is incorrect about Flink's implementation, > >> please correct me. > >> > >> On the other hand, if we want to do symmetric merge in Kafka Streams, we > >> can change the window definition to [start time - gap, start time + > gap]. > >> This way the example (7,2),(10, 10), (19,5),(15,3) will be > >> [5,9] > >> [0,20] => merged to [0,20] > >> [14,24] => merged to [0,24] > >> [12,18] => merged to [0,24] > >> > >> (19,5),(15,3)(7,2),(10, 10) will generate same result > >> [14,24] > >> [12,18] => merged to [12,24] > >> [5,9] => no merge > >> [0,20] => merged to [0,24] > >> > >> Note that symmetric-merge would require us to change the way how Kafka > >> Steams fetch windows now, instead of fetching range from timestamp-gap > to > >> timestamp+gap, we will need to fetch all windows that are not expired > yet. > >> On the other hand, I'm not sure how this will impact the current logic > of > >> how a window is considered as closed, because the window doesn't carry > end > >> timestamp anymore, but end timestamp + gap. > >> > >> So do you guys think forward-merge approach used by Flink makes more > sense > >> in Kafka Streams, or symmetric-merge makes more sense? Both of them > seems > >> to me can give deterministic result. > >> > >> BTW I'll add the use case into original KIP. > >> > >> Lei > >> > >> > >> On Tue, Sep 11, 2018 at 5:45 PM Matthias J. Sax <matth...@confluent.io> > >> wrote: > >> > >>> Thanks for explaining your understanding. And thanks for providing more > >>> details about the use-case. Maybe you can add this to the KIP? > >>> > >>> > >>> First one general comment. I guess that my and Guozhangs understanding > >>> about gap/close/gracePeriod is the same as yours -- we might not have > >>> use the term precisely correct in previous email. > >>> > >>> > >>> To you semantics of gap in detail: > >>> > >>>> I thought when (15,3) is received, kafka streams look up for neighbor > >>>> record/window that is within the gap > >>>> of [15-3, 15+3], and merge if any. Previous record (10, 10) created > its > >>> own > >>>> window [10, 10], which is > >>>> out of the gap, so nothing will be found and no merge occurs. Hence we > >>> have > >>>> two windows now in session store, > >>>> [10, 10] and [15, 15] respectively. > >>> > >>> If you have record (10,10), we currently create a window of size > >>> [10,10]. When record (15,3) arrives, your observation that the gap 3 is > >>> too small to be merged into [10,10] window -- however, merging is a > >>> symmetric operation and the existing window of [10,10] has a gap of 10 > >>> defined: thus, 15 is close enough to fall into the gap, and (15,3) is > >>> merged into the existing window resulting in window [10,15]. > >>> > >>> If we don't respect the gap both ways, we end up with inconsistencies > if > >>> data is out-of-order. For example, if we use the same input record > >>> (10,10) and (15,3) from above, and it happens that (15,3) is processed > >>> first, when processing out-of-order record (10,10) we would want to > >>> merge both into a single window, too? > >>> > >>> Does this make sense? > >>> > >>> Now the question remains, if two records with different gap parameter > >>> are merged, which gap should we apply for processing/merging future > >>> records into the window? It seems, that we should use the gap parameter > >>> from the record with this larges timestamp. In the example above > (15,3). > >>> We would use gap 3 after merging independent of the order of > processing. > >>> > >>> > >>>> Also another thing worth mentioning is that, the session window object > >>>> created in current kafka streams > >>>> implementation doesn't have gap info, it has start and end, which is > the > >>>> earliest and latest event timestamp > >>>> in that window interval, i.e for (10,10), the session window gets > created > >>>> is [10,10], rather than [10,20]. Just to clarify > >>>> so that it's clear why (10,10) cannot be fetched when looking for gap > of > >>>> (15,3), it's because the end boundary 10 of > >>>> [10,10] is smaller than search boundary [12,18]. > >>> > >>> We don't need to store the gap, because the gap is know from the window > >>> definition. The created window size depends on the data that is > >>> contained in the window. I guess one could define it differently, too, > >>> ie, for the (10,10) record, we create a window [0,20] -- not sure if it > >>> makes a big difference in practice though. Note, that creating window > >>> [10,20] would not be correct, because the gap must be applied in both > >>> directions, not just into the future. > >>> > >>> About the second part: the search would not be applied from (15,3) in > >>> range [12,18], but from existing window [10,10] into range [0,20] and > 15 > >>> is contained there. This example also shows, that we would need to come > >>> up with a clever way, to identify window [10,10] when processing (15,3) > >>> -- not sure atm how to do this. However, only consider (15,3) would > >>> result in inconsistencies for out-of-order data as pointed out above > and > >>> would not be sufficient. > >>> > >>> > >>> Does this make sense? > >>> > >>> > >>> Or is there another way to define dynamic session gap semantics in a > >>> deterministic way with regard to out-of-order data? > >>> > >>> > >>> > >>> -Matthias > >>> > >>> > >>> On 9/11/18 4:28 PM, Lei Chen wrote: > >>>> Thanks Matthias and Guozhang for the response. > >>>> > >>>> Seems like our understanding mainly differs in the semantics of gap in > >>>> session windows. > >>>> > >>>> My understanding is that gap is used to merge nearby records together > >>> such > >>>> that no record > >>>> in the merged window has distance later than gap. In Kafka Streams's > >>>> implementation it's > >>>> mainly used to find neighbor records/windows in session store so that > >>>> nearby records can > >>>> be merge. It is NOT used to determine when a window should be closed, > >>> which > >>>> is in > >>>> fact determined by window's grace period. > >>>> > >>>> Guozhang you said "b. When later we received (15, 3), it means that > this > >>>> record ** changed ** > >>>> the window gap interval from 10 to 3, and hence we received a new > record > >>> at > >>>> 15, with the new window gap of 3, it means that by timestamp 18 (15 + > 3) > >>> if > >>>> we have not received any new data, the window should be closed, i.e. > the > >>>> window is now [10, 18) which includes two records at 10 and 15." > >>>> > >>>> This is different from what i thought will happen. > >>>> > >>>> I thought when (15,3) is received, kafka streams look up for neighbor > >>>> record/window that is within the gap > >>>> of [15-3, 15+3], and merge if any. Previous record (10, 10) created > its > >>> own > >>>> window [10, 10], which is > >>>> out of the gap, so nothing will be found and no merge occurs. Hence we > >>> have > >>>> two windows now in session store, > >>>> [10, 10] and [15, 15] respectively. > >>>> > >>>> Also another thing worth mentioning is that, the session window object > >>>> created in current kafka streams > >>>> implementation doesn't have gap info, it has start and end, which is > the > >>>> earliest and latest event timestamp > >>>> in that window interval, i.e for (10,10), the session window gets > created > >>>> is [10,10], rather than [10,20]. Just to clarify > >>>> so that it's clear why (10,10) cannot be fetched when looking for gap > of > >>>> (15,3), it's because the end boundary 10 of > >>>> [10,10] is smaller than search boundary [12,18]. > >>>> > >>>> Please correct me if my understanding is wrong here. > >>>> > >>>> @Matthias, to answer your use case question, we have an use case where > >>>> asynchronous time series data > >>>> are received in the stream, from different contributors, with > different > >>>> quality and at different pace. > >>>> Inside Kafka Streams, we use state to maintain statistic aggregations > and > >>>> other mathematics model to track > >>>> the liquidity and calculate time decay rate and dynamic gap, so that > at > >>>> runtime, for each contributor we can > >>>> 1. determine how many historical records we should maintain in state. > >>>> 2. for each incoming record, output a record using aggregations from > >>>> *nearby* records from that contributor. > >>>> Why fixed gap session window doesn't work here? Because the > definition of > >>>> "nearby" here is determined by > >>>> several very dynamic factors in our case, it changes not only with > >>>> different hours in a day, but also related to > >>>> other contributors. > >>>> The purpose of this KIP is to suggest a dynamic session window > >>>> implementation so that we can embed such > >>>> dynamic "nearby" calculation capability into kafka streams session > >>> windows > >>>> semantics. Hope it makes sense to you. > >>>> > >>>> Lei > >>>> > >>>> > >>>> > >>>> > >>>> On Mon, Sep 10, 2018 at 5:27 PM Guozhang Wang <wangg...@gmail.com> > >>> wrote: > >>>> > >>>>> Hello Lei, > >>>>> > >>>>> As Matthias mentioned, the key question here is that because of the > late > >>>>> arrivals of records which may indicate a shorter session gap > interval, > >>> some > >>>>> session windows may be "mistakenly" merged and hence need to be > undone > >>> the > >>>>> merge, i.e. to split them again. > >>>>> > >>>>> Back to my example, you are right that the processing result of > >>>>> > >>>>> (10, 10), (19, 5), (15, 3) .. > >>>>> > >>>>> should be the same as the processing result of > >>>>> > >>>>> (10, 10), (15, 3), (19, 5) .. > >>>>> > >>>>> Note that the second value is NOT the window end time, but the > extracted > >>>>> window gap interval, as you suggested in the KIP this value can be > >>>>> dynamically changed > >>>>> > >>>>> a. If you take a look at the second ordering, when we receive (10, > 10) > >>> it > >>>>> means a window starting at 10 is created, and its gap interval is 10, > >>> which > >>>>> means that if by the timestamp of 20 we do not receive any new data, > >>> then > >>>>> the window should be closed, i.e. the window [10, 20). > >>>>> > >>>>> b. When later we received (15, 3), it means that this record ** > changed > >>> ** > >>>>> the window gap interval from 10 to 3, and hence we received a new > >>> record at > >>>>> 15, with the new window gap of 3, it means that by timestamp 18 (15 + > >>> 3) if > >>>>> we have not received any new data, the window should be closed, i.e. > the > >>>>> window is now [10, 18) which includes two records at 10 and 15. > >>>>> > >>>>> c. The third record is received at 19, which is after the window > close > >>> time > >>>>> 18, it means that we should now start a new window starting at 19, > i.e. > >>> the > >>>>> window is [19, 24), > >>>>> > >>>>> > >>>>> BUT, because of the out of ordering, we did not receive (15, 3) in > time, > >>>>> but received (19, 5), it will cause us to mistakenly merge the > window of > >>>>> [10, 20) with [19, 24) to [10, 24), and only when later we received > >>> (15, 3) > >>>>> we realized that the previous window should have been ended at 18. > >>>>> > >>>>> Does that make sense to you? > >>>>> > >>>>> > >>>>> Guozhang > >>>>> > >>>>> > >>>>> On Thu, Sep 6, 2018 at 9:50 PM, Matthias J. Sax < > matth...@confluent.io> > >>>>> wrote: > >>>>> > >>>>>> I cannot follow the example: > >>>>>> > >>>>>>>> (10, 10), (15, 3), (19, 5) ... > >>>>>> > >>>>>> First, [10,10] is created, second the window is extended to [10,15], > >>> and > >>>>>> third [19,19] is created. Why would there be a [15,15]? And why > would > >>>>>> (19,5) be merged into [15,15] -- the gap was set to 3 via (15,3) and > >>>>>> thus [19,19] should be its own window? > >>>>>> > >>>>>>> Take a look at another example, > >>>>>>> (13, 3), (19, 5), (15, 3) ... > >>>>>>> > >>>>>>> in this case when (15, 3) is received, [13,13] should be retrieved > and > >>>>>>> merged to a new window [13, 15], then [19,19] should be updated to > >>> [13, > >>>>>>> 19]. Correct? > >>>>>> > >>>>>> This example makes sense. However, Guozhang's example was different. > >>> The > >>>>>> late even, _reduces_ the gap and this can lead to a window split. > >>>>>> Guozhang's example was > >>>>>> > >>>>>>>>> (10, 10), (19, 5), (15, 3) ... > >>>>>> > >>>>>> First [10,10] is created, second [10,19] is create (gap is 10, so 10 > >>> and > >>>>>> 19 merge). Last, (15,3) reduced the gap from 10 to 3, thus [10,15] > and > >>>>>> [19,19] must be two windows, ie, original window [10,19] must be > split. > >>>>>> > >>>>>> > >>>>>> Or maybe you have different semantic about gaps are dynamically > >>> modified > >>>>>> in mind? It's a little unclear for the KIP itself what semantics > >>> dynamic > >>>>>> sessions windows should have. > >>>>>> > >>>>>> > >>>>>> What is also unclear to me atm is, what use cases you have in mind? > The > >>>>>> KIP only says > >>>>>> > >>>>>>> the statistical aggregation result, liquidity of the records, > >>>>>> > >>>>>> > >>>>>> I am not sure what this means. Can you elaborate? > >>>>>> > >>>>>> > >>>>>> > >>>>>> -Matthias > >>>>>> > >>>>>> > >>>>>> > >>>>>> On 8/30/18 3:32 PM, Lei Chen wrote: > >>>>>>> Hi Guozhang, > >>>>>>> > >>>>>>> Thanks for reviewing the proposal. I didn't think of out of order > >>>>> events > >>>>>>> and glad that you brought it up. > >>>>>>> > >>>>>>> In the example you gave, > >>>>>>> > >>>>>>> (10, 10), (19, 5), (15, 3) ... > >>>>>>> > >>>>>>> my understanding is that the correct result window should be the > same > >>>>> as > >>>>>> in > >>>>>>> order events > >>>>>>> > >>>>>>> (10, 10), (15, 3), (19, 5) ... > >>>>>>> > >>>>>>> when (15, 3) is received, [15,15] is creatd > >>>>>>> when (19, 5) is received, [15, 15] and [19, 19] are merged and [15, > >>> 19] > >>>>>> is > >>>>>>> created, meanwhile [15,15] is removed > >>>>>>> > >>>>>>> back to out of order case, > >>>>>>> > >>>>>>> when (19 ,5) is received, [19, 19] is created > >>>>>>> when (15, 3) is received, in order to generate the same result, > >>>>>>> 1. if late event is later than retention period, it will be dropped > >>>>>>> 2. otherwise, adjacent session windows within gap should be > retrieved > >>>>> and > >>>>>>> merged accordingly, in this case [19, 19], and create a new session > >>>>> [15, > >>>>>> 19] > >>>>>>> I'm little confused when you said "the window [15, 15] SHOULD > actually > >>>>> be > >>>>>>> expired at 18 and hence the next record (19, 5) should be for a new > >>>>>> session > >>>>>>> already.". If i understand it correctly, the expiration of the > window > >>>>> is > >>>>>>> only checked when next event (19,5) comes and then it should be > merged > >>>>> to > >>>>>>> it. [15, 15] will then be closed. Is that also what you meant? > >>>>>>> I cannot think of a case where a window will be split by a late > event, > >>>>>>> because if event A and C fall into the same session window, a late > >>>>> event > >>>>>> B > >>>>>>> in middle will definitely fall into C's gap as well. IOW, late > event > >>>>> will > >>>>>>> only cause window extension, not split. > >>>>>>> > >>>>>>> Take a look at another example, > >>>>>>> (13, 3), (19, 5), (15, 3) ... > >>>>>>> > >>>>>>> in this case when (15, 3) is received, [13,13] should be retrieved > and > >>>>>>> merged to a new window [13, 15], then [19,19] should be updated to > >>> [13, > >>>>>>> 19]. Correct? > >>>>>>> > >>>>>>> To be able to achieve that, like you said, the gap needs to be > stored > >>>>> for > >>>>>>> sessions. We don't need to save the gap with each event, but only > for > >>>>>> each > >>>>>>> session window. To avoid upgrading existing session window, how > about > >>>>>>> create a new Window type extended from SessionWindow along with a > new > >>>>>>> KeySchema? > >>>>>>> > >>>>>>> What do you think? > >>>>>>> > >>>>>>> Lei > >>>>>>> > >>>>>>> > >>>>>>> On Fri, Aug 24, 2018 at 9:42 AM Guozhang Wang <wangg...@gmail.com> > >>>>>> wrote: > >>>>>>> > >>>>>>>> Hello Lei, > >>>>>>>> > >>>>>>>> Thanks for the proposal. I've just made a quick pass over it and > >>> there > >>>>>> is a > >>>>>>>> question I have: > >>>>>>>> > >>>>>>>> The session windows are defined per key, i.e. does that mean that > >>> each > >>>>>>>> incoming record of the key can dynamically change the gap of the > >>>>> window? > >>>>>>>> For example, say you have the following record for the same key > >>> coming > >>>>>> in > >>>>>>>> order, where the first time is the timestamp of the record, and > the > >>>>>> second > >>>>>>>> value is the extracted gap value: > >>>>>>>> > >>>>>>>> (10, 10), (19, 5), ... > >>>>>>>> > >>>>>>>> > >>>>>>>> When we receive the first record at time 10, the gap is extracted > as > >>>>> 10, > >>>>>>>> and hence the window will be expired at 20 if no other record is > >>>>>> received. > >>>>>>>> When we receive the second record at time 19, the gap is modified > to > >>>>> 5, > >>>>>> and > >>>>>>>> hence the window will be expired at 24 if no other record is > >>> received. > >>>>>>>> > >>>>>>>> > >>>>>>>> If that's the case, I'm wondering how out-of-order data can be > >>> handled > >>>>>>>> then, consider this stream: > >>>>>>>> > >>>>>>>> (10, 10), (19, 5), (15, 3) ... > >>>>>>>> > >>>>>>>> I.e. you received a late record indicating at timestamp 15, which > >>>>>> shorten > >>>>>>>> the gap to 3. It means that the window SHOULD actually be expired > at > >>>>> 18, > >>>>>>>> and hence the next record (19, 5) should be for a new session > >>> already. > >>>>>>>> Today Streams session window implementation does not do "window > >>>>> split", > >>>>>> so > >>>>>>>> have you thought about how this can be extended? > >>>>>>>> > >>>>>>>> Also since in your proposal each session window's gap value would > be > >>>>>>>> different, we need to store this value along with each record > then, > >>>>> how > >>>>>>>> would we store it, and what would be the upgrade path if it is > not a > >>>>>>>> compatible change on disk storage etc? > >>>>>>>> > >>>>>>>> > >>>>>>>> > >>>>>>>> Guozhang > >>>>>>>> > >>>>>>>> > >>>>>>>> > >>>>>>>> On Wed, Aug 22, 2018 at 10:05 AM, Lei Chen <ley...@gmail.com> > wrote: > >>>>>>>> > >>>>>>>>> Hi All, > >>>>>>>>> > >>>>>>>>> I created a KIP to add dynamic gap session window support to > Kafka > >>>>>>>> Streams > >>>>>>>>> DSL. > >>>>>>>>> > >>>>>>>>> https://cwiki.apache.org/confluence/display/KAFKA/KIP- > >>>>>>>>> 362%3A+Support+dynamic+gap+session+window > >>>>>>>>> > >>>>>>>>> Please take a look, > >>>>>>>>> > >>>>>>>>> Thanks, > >>>>>>>>> Lei > >>>>>>>>> > >>>>>>>> > >>>>>>>> > >>>>>>>> > >>>>>>>> -- > >>>>>>>> -- Guozhang > >>>>>>>> > >>>>>>> > >>>>>> > >>>>>> > >>>>> > >>>>> > >>>>> -- > >>>>> -- Guozhang > >>>>> > >>>> > >>> > >>> > >> > > > >
