Hi Garrett,
I'm glad this helped.
You're absolutely right, only the "oneMinuteWindowed" KTable has a
Windowed key - apologies again for getting it wrong the first time.
I admit I used window().end() arbitrarily. If window().start() matches
your semantics better, use that. Further on that note (I didn't propose
this originally simply because it is more error-prone for me to write)
you can instead of Tuple2 actually use the Windowed class as the wrapper
and then have both start and end time if you have use for them
downstream. Then e.g. fiveMinuteWindowed becomes (notice Windowed is
used intentionally this time!):
KTable<Windowed<String>, CountSumMinMaxAvgObj> fiveMinuteWindowed =
oneMinuteWindowed
.groupBy( (windowedKey, value) ->
new KeyValue<>(
new Windowed<>(
windowedKey.key(),
new Window<>(
windowedKey.window().start() /1000/60/5 *1000*60*5,
windowedKey.window().start() /1000/60/5 *1000*60*5 + 1000*60*5
)
),
value,new SessionKeySerde<>(Serdes.String()), valSerde)
.reduce(/*your adder*/, /*your subtractor*/, "store5m");
In the above, regardless if you choose to use start() or end(), you need
to use it consistently both times, as in the time-windowed aggregation
on your original stream the 1-minute window start and end won't be
aligned to full minutes (I assume), hence they could end up in two
different 5-minute windows after rounding.
As to the generic types, in my experience getting well-typed Serdes gets
the type parameters inferred correctly throughout the rest of the
expression and they should only be needed on the KTable variable
declaration.
And if you use Windowed keys as in the variant above, you can (sneakily)
re-use the SessionKeySerde as a wrapper around the actual serde for your
key, like this:
new SessionKeySerde<>(Serdes.String())
Again this should get type parameters inferred nicely and
SessionKeySerde serializes both start and end time of the window.
*Caveat*: SessionKeySerde comes from the
org.apache.kafka.streams.kstream.*internals* package, so it's not as
safe to use as public APIs, but it's a small class you can fork if you wish.
This is so much fun ;-)
Good luck,
Michał
On 08/05/17 23:05, Garrett Barton wrote:
Michael,
This is slick! I am still writing unit tests to verify it. My code
looks something like:
KTable<Windowed<String>, CountSumMinMaxAvgObj> oneMinuteWindowed =
srcStream // my val object isnt really called that, just wanted to show
a sample set of calculations the value can do!
.groupByKey(Serdes.String(), Serdes.Double())
.aggregate(/*initializer */, /* aggregator */, TimeWindows.of(60*1000,
60*1000), "store1m");
// i used an aggregate here so I could have a non-primitive value object
that does the calculations on each aggregator, pojo has an .add(Double) in
it.
KTable<Tuple2<String, Long>, CountSumMinMaxAvgObj> fiveMinuteWindowed =
oneMinuteWindowed // I made my own Tuple2, will move window calc into it
.groupBy( (windowedKey, value) -> new KeyValue<>(new Tuple2<String,
Long>(windowedKey.key(), windowedKey.window().end() /1000/60/5 *1000*60*5),
value, keySerde, valSerde)
// the above rounds time down to a timestamp divisible by 5 minutes
.reduce(/*your adder*/, /*your subtractor*/, "store5m");
// where your subtractor can be as simple as (val, agg) -> agg - val
for primitive types or as complex as you need,
// just make sure you get the order right (lesson hard learnt ;) ),
subtraction is not commutative!
// again my val object has an .add(Obj) and a .sub() to handle
this, so nice!
KTable<Tuple2<String, Long>, CountSumMinMaxAvgObj> fifteenMinuteWindowed =
fiveMinuteWindowed
.groupBy( (keyPair, value) -> new KeyValue<>(new Tuple2(keyPair._1,
keyPair._2 /1000/60/15 *1000*60*15), value, keySerde, valSerde)
// the above rounds time down to a timestamp divisible by 15 minutes
.reduce(/*your adder*/, /*your subtractor*/, "store15m");
KTable<Tuple2<String, Long>, CountSumMinMaxAvgObj> sixtyMinuteWindowed =
fifteeenMinuteWindowed
.groupBy( (keyPair, value) -> new KeyValue<>(new Tuple2(keyPairair._1,
pair._2 /1000/60/60 *1000*60*60), value, keySerde, valSerde)
// the above rounds time down to a timestamp divisible by 60 minutes
.reduce(/*your adder*/, /*your subtractor*/, "store60m");
Notes thus far:
Doesn't look like I need to start the 5min with a windowed KTable return
object, it starts with the regular KTable<Tuple2<String,Long>> in this case.
I thinking about using windowedKey.window().start() instead of end() as I
believe that is more consistent with what the windows themselves put out.
They go into the stores bound by their start time I believe.
Serdes gets nuts as well as the Generic typing on some of these classes
(yea you KeyValueMapper), makes for long code! I had to specify them
everywhere since the key/val's changed.
I didn't get enough time to mess with it today, I will wrap up the unit
tests and run it to see how it performs against my real data as well
tomorrow. I expect a huge reduction in resources (both streams and kafka
storage) by moving to this.
Thank you!
On Mon, May 8, 2017 at 5:26 PM, Matthias J. Sax <matth...@confluent.io>
wrote:
Michal,
that's an interesting idea. In an ideal world, Kafka Streams should have
an optimizer that is able to to this automatically under the hood. Too
bad we are not there yet.
@Garret: did you try this out?
This seems to be a question that might affect many users, and it might
we worth to document it somewhere as a recommended pattern.
-Matthias
On 5/8/17 1:43 AM, Michal Borowiecki wrote:
Apologies,
In the code snippet of course only oneMinuteWindowed KTable will have a
Windowed key (KTable<Windowed<Key>, Value>), all others would be just
KTable<Tuple2<Key, Long>, Value>.
Michał
On 07/05/17 16:09, Michal Borowiecki wrote:
Hi Garrett,
I've encountered a similar challenge in a project I'm working on (it's
still work in progress, so please take my suggestions with a grain of
salt).
Yes, I believe KTable.groupBy lets you accomplish what you are aiming
for with something like the following (same snippet attached as txt
file):
KTable<Windowed<Key>, Value> oneMinuteWindowed = yourStream //
where Key and Value stand for your actual key and value types
.groupByKey()
.reduce(/*your adder*/, TimeWindows.of(60*1000, 60*1000),
"store1m");
//where your adder can be as simple as (val, agg) -> agg + val
//for primitive types or as complex as you need
KTable<Windowed<Tuple2<Key, Long>>, Value> fiveMinuteWindowed =
oneMinuteWindowed // Tuple2 for this example as defined by
javaslang library
.groupBy( (windowedKey, value) -> new KeyValue<>(new
Tuple2<>(windowedKey.key(), windowedKey.window().end() /1000/60/5
*1000*60*5), value)
// the above rounds time down to a timestamp divisible by 5
minutes
.reduce(/*your adder*/, /*your subtractor*/, "store5m");
// where your subtractor can be as simple as (val, agg) -> agg
- valfor primitive types or as complex as you need,
// just make sure you get the order right (lesson hard learnt
;) ), subtraction is not commutative!
KTable<Windowed<Tuple2<Key, Long>>, Value> fifteenMinuteWindowed =
fiveMinuteWindowed
.groupBy( (keyPair, value) -> new KeyValue<>(new
Tuple2(keyPair._1, keyPair._2/1000/60/15 *1000*60*15), value)
// the above rounds time down to a timestamp divisible by 15
minutes
.reduce(/*your adder*/, /*your subtractor*/, "store15m");
KTable<Windowed<Tuple2<Key, Long>>, Value> sixtyMinuteWindowed =
fifteeenMinuteWindowed
.groupBy( (keyPair, value) -> new KeyValue<>(new
Tuple2(keyPairair._1, pair._2 /1000/60/60 *1000*60*60), value)
// the above rounds time down to a timestamp divisible by 5
minutes
.reduce(/*your adder*/, /*your subtractor*/, "store60m");
So, step by step:
* You use a windowed aggregation only once, from there on you use
the KTable abstraction only (which doesn't have windowed
aggregations).
* In each subsequent groupBy you map the key to a pair of
(your-real-key, timestamp) where the timestamp is rounded down
with the precision of the size of the new window.
* reduce() on a KGroupedTable takes an adder and a subtractor and it
will correctly update the new aggregate by first subtracting the
previous value of the upstream record before adding the new value
(this way, just as you said, the downstream is aware of the
statefulness of the upstream and correctly treats each record as
an update)
* If you want to reduce message volume further, you can break these
into separate KafkaStreams instances and configure downstream ones
with a higher commit.interval.ms (unfortunately you can't have
different values of this setting in different places of the same
topology I'm afraid)
* TODO: Look into retention policies, I haven't investigated that in
any detail.
I haven't tested this exact code, so please excuse any typos.
Also, if someone with more experience could chip in and check if I'm
not talking nonsense here, or if there's an easier way to this, that
would be great.
I don't know if the alternative approach is possible, where you
convert each resulting KTable back into a stream and just do a
windowed aggregation somehow. That would feel more natural, but I
haven't figured out how to correctly window over a changelog in the
KStream abstraction, feels impossible in the high-level DSL.
Hope that helps,
Michal
On 02/05/17 18:03, Garrett Barton wrote:
Lets say I want to sum values over increasing window sizes of 1,5,15,60
minutes. Right now I have them running in parallel, meaning if I am
producing 1k/sec records I am consuming 4k/sec to feed each
calculation.
In reality I am calculating far more than sum, and in this pattern I'm
looking at something like (producing rate)*(calculations)*(windows)
for a
consumption rate.
So I had the idea, could I feed the 1 minute window into the 5
minute, and
5 into 15, and 15 into 60. Theoretically I would consume a fraction
of the
records, not have to scale as huge and be back to something like
(producing
rate)*(calculations)+(updates).
Thinking this is an awesome idea I went to try and implement it and
got
twisted around. These are windowed grouping operations that produce
KTables, which means instead of a raw stream I have an update stream.
To
me this implies that downstream must be aware of this and consume
stateful
information, knowing that each record is an update and not an in
addition
to. Does the high level api handle that construct and let me do
that? For
a simple sum it would have to hold each of the latest values for say
the 5
1 minute sum's in a given window, to perform the 5 minute sum.
Reading the
docs which are awesome, I cannot determine if the KTable.groupby()
would
work over a window, and would reduce or aggregate thus do what I need?
Any ideas?
--
Signature
<http://www.openbet.com/> Michal Borowiecki
Senior Software Engineer L4
T: +44 208 742 1600
+44 203 249 8448
E: michal.borowie...@openbet.com
W: www.openbet.com <http://www.openbet.com/>
OpenBet Ltd
Chiswick Park Building 9
566 Chiswick High Rd
London
W4 5XT
UK
<https://www.openbet.com/email_promo>
This message is confidential and intended only for the addressee. If
you have received this message in error, please immediately notify the
postmas...@openbet.com <mailto:postmas...@openbet.com> and delete it
from your system as well as any copies. The content of e-mails as well
as traffic data may be monitored by OpenBet for employment and
security purposes. To protect the environment please do not print this
e-mail unless necessary. OpenBet Ltd. Registered Office: Chiswick Park
Building 9, 566 Chiswick High Road, London, W4 5XT, United Kingdom. A
company registered in England and Wales. Registered no. 3134634. VAT
no. GB927523612
--
Signature
<http://www.openbet.com/> Michal Borowiecki
Senior Software Engineer L4
T: +44 208 742 1600
+44 203 249 8448
E: michal.borowie...@openbet.com
W: www.openbet.com <http://www.openbet.com/>
OpenBet Ltd
Chiswick Park Building 9
566 Chiswick High Rd
London
W4 5XT
UK
<https://www.openbet.com/email_promo>
This message is confidential and intended only for the addressee. If you
have received this message in error, please immediately notify the
postmas...@openbet.com <mailto:postmas...@openbet.com> and delete it
from your system as well as any copies. The content of e-mails as well
as traffic data may be monitored by OpenBet for employment and security
purposes. To protect the environment please do not print this e-mail
unless necessary. OpenBet Ltd. Registered Office: Chiswick Park Building
9, 566 Chiswick High Road, London, W4 5XT, United Kingdom. A company
registered in England and Wales. Registered no. 3134634. VAT no.
GB927523612
