I just want to add something: If I understand the question correctly, you are asking for a strong ordering guarantee. I personally doubt that out-of-order on count-based windows can be supported with strong consistency guarantee in an efficient manner.
If a late record arrives for window X, the newest record from window X
must be removed and also added to the next window (which starts a
cascaded update of all window up to "now" -- at least for
non-overlapping windows).
Let the input be A,B,C,D,E,F,G,... and let C be the late record. Thus,
you would initially build (eg, 5-tuple) tumbling windows as follows:
ABDEF GHIJK LMNOP
On late arriving of C, you update as follows
insert C in W1
move F from W1 to W2
move K from W2 to W3
(and so on up to "now")
=> ABCDE FGHIJ KLMNOP
This can be quite expensive, depending on the time lag of C.
For a window hop of one tuple the cascading update can be terminated
earlier. (For other hop values, you should think careful to get it right).
Let the input be A,B,C,D,E,F,G,... and let C be the late record. Thus,
you would initially build the following windows:
ABDEF
BDEFG
DEFGH
On late arriving of C, you update as follows
ABDEF => ABCDE (remove F, add C)
BDEFG => BCDEF (remove G, add C)
=> CDEFG (new window; terminates the update sequence)
DEFGH => DEFGH (stay the same)
Hope this helps to understand the issue...
-Matthias
On 06/24/2016 12:21 AM, Guozhang Wang wrote:
> Hello Ryan,
>
> On the DSL layer, currently there is not support for record window yet; and
> we are discussing about adding such support in the future, maybe first
> session windows then others.
>
> On the Processor API layer, you can definitely implement this "record
> window" feature yourself by keeping track of the most recent 10 records in
> a state store, and upon each new incoming record, delete the oldest record
> and insert this new record, and re-compute the count (for your specific
> case you do not need to re-scan all 10 records, but just get the new counts
> as
>
> old counts + (if new record > 100) ? 1 : 0 - (if old record > 100) ? 1 : 0
>
>
> Guozhang
>
>
> On Thu, Jun 23, 2016 at 1:55 PM, Ryan Thompson <[email protected]>
> wrote:
>
>> Hello,
>>
>> Say I have a stream, and want to determine whether or not a given "density"
>> of of records match a given condition. For example, let's say I want to
>> how many of the last 10 records have a numerical value greater than 100.
>>
>> Does the kafka streams DSL (or processor API) provide a way to do this type
>> of aggregation in a way that supports out of order messages?
>>
>> I can't use a time window based aggregation here because my window is based
>> on a quantity of records (in this case, the last 10) rather than time.
>> However, I still want to know about the last 10 records regardless of what
>> order they arrive.
>>
>> Thanks,
>> Ryan
>>
>
>
>
signature.asc
Description: OpenPGP digital signature
