If you set the key to the time attribute, the "old" key is no longer valid. The streams are organized by time and only one aggregate for each window-time should be computed.
This should do what you are looking for: DataStream .keyBy(_._1) // key by orginal key .timeWindow(..) .apply(...) // extract window end time: (origKey, time, agg) .keyBy(_._2) // key by time field .maxBy(_._3) // value with max agg field Best, Fabian 2015-11-23 11:00 GMT+01:00 Konstantin Knauf <[email protected]>: > Hi Fabian, > > thanks for your answer. Yes, that's what I want. > > The solution you suggest is what I am doing right now (see last of the > bullet point in my question). > > But given your example. I would expect the following output: > > (key: 1, w-time: 10, agg: 17) > (key: 2, w-time: 10, agg: 20) > (key: 1, w-time: 20, agg: 30) > (key: 1, w-time: 20, agg: 30) > (key: 1, w-time: 20, agg: 30) > > Because the reduce function is evaluated for every incoming event (i.e. > each key), right? > > Cheers, > > Konstantin > > On 23.11.2015 10:47, Fabian Hueske wrote: > > Hi Konstantin, > > > > let me first summarize to make sure I understood what you are looking > for. > > You computed an aggregate over a keyed event-time window and you are > > looking for the maximum aggregate for each group of windows over the > > same period of time. > > So if you have > > (key: 1, w-time: 10, agg: 17) > > (key: 2, w-time: 10, agg: 20) > > (key: 1, w-time: 20, agg: 30) > > (key: 2, w-time: 20, agg: 28) > > (key: 3, w-time: 20, agg: 5) > > > > you would like to get: > > (key: 2, w-time: 10, agg: 20) > > (key: 1, w-time: 20, agg: 30) > > > > If this is correct, you can do this as follows. > > You can extract the window start and end time from the TimeWindow > > parameter of the WindowFunction and key the stream either by start or > > end time and apply a ReduceFunction on the keyed stream. > > > > Best, Fabian > > > > 2015-11-23 8:41 GMT+01:00 Konstantin Knauf <[email protected] > > <mailto:[email protected]>>: > > > > Hi everyone, > > > > me again :) Let's say you have a stream, and for every window and key > > you compute some aggregate value, like this: > > > > DataStream.keyBy(..) > > .timeWindow(..) > > .apply(...) > > > > > > Now I want to get the maximum aggregate value for every window over > the > > keys. This feels like a pretty natural use case. How can I achieve > this > > with Flink in the most compact way? > > > > The options I thought of so far are: > > > > * Use an allTimeWindow, obviously. Drawback is, that the > WindowFunction > > would not be distributed by keys anymore. > > > > * use a windowAll after the WindowFunction to create windows of the > > aggregates, which originated from the same timeWindow. This could be > > done either with a TimeWindow or with a GlobalWindow with > DeltaTrigger. > > Drawback: Seems unnecessarily complicated and doubles the latency (at > > least in my naive implementation ;)). > > > > * Of course, you could also just keyBy the start time of the window > > after the WindowFunction, but then you get more than one event for > each > > window. > > > > Is there some easy way I am missing? If not, is there a technical > > reasons, why such an "reduceByKeyAndWindow"-operator is not > available in > > Flink? > > > > Cheers, > > > > Konstantin > > > > > > -- > Konstantin Knauf * [email protected] * +49-174-3413182 > TNG Technology Consulting GmbH, Betastr. 13a, 85774 Unterföhring > Geschäftsführer: Henrik Klagges, Christoph Stock, Dr. Robert Dahlke > Sitz: Unterföhring * Amtsgericht München * HRB 135082 >
