AW: max aggregator dosen't work as expected

[Claudia Wegmann]

So if I change the input data to (I added an uID value to identify the single 
data sets):

        new Data(1, 123, new Date(116, 8,8,11,11,11), 5),

        new Data(2, 123, new Date(116, 8,8,12,10,11), 8),

        new Data(3, 123, new Date(116, 8,8,12,12,11), 10),

        new Data(4, 123, new Date(116, 8,8,12,15,11), 2))

the expected output for maxBy would be:

       Data(1, 123, Date(116, 8,8,12,10,11), 8),

       Data(3, 123, Date(116, 8,8,12,12,11), 10),
and for max would be:

       Data(null, null, null, 8),

       Data(null, null, null, 10),
?

My output from Flink is:
1> Data: uID-1|| ID-123 || timestamp-Thu Sep 08 11:11:11 CEST 2016 || 
temperatur-5
1> Data: uID-2|| ID-123 || timestamp-Thu Sep 08 12:10:11 CEST 2016 || 
temperatur-10

Here are the new questions:

1.)    So, does max not work as expected?


2.)    And doesn’t the time window stat with the first incoming record? 
Obviously first and second data set are not in the same window. When does the 
time window start?


3.)    Why does the second record in the result have uID = 2 although it had 
uID = 3 in the input data set?

Thanks again and best,
Claudia

Von: Robert Metzger [mailto:rmetz...@apache.org]
Gesendet: Montag, 8. August 2016 15:52
An: user@flink.apache.org
Betreff: Re: max aggregator dosen't work as expected

I have to admit that the difference between the two methods is subtle, and in 
my opinion it doesn't make much sense to have the two variants.

- max() returns a tuple with the max value at the specified position, the other 
fields of the tuple/pojo are undefined
- maxBy() returns a tuple with the max value at the specified position (the 
other fields are retained)

On Mon, Aug 8, 2016 at 2:55 PM, Claudia Wegmann 
<c.wegm...@kasasi.de<mailto:c.wegm...@kasasi.de>> wrote:
OK, found my mistake reagarding question 2.). I key by the id value and gave 
all the data sets different values there. So of course all 4 data sets are 
printed. Sorry ☺
But question 1.) still remains.

Von: Claudia Wegmann [mailto:c.wegm...@kasasi.de<mailto:c.wegm...@kasasi.de>]
Gesendet: Montag, 8. August 2016 14:27
An: user@flink.apache.org<mailto:user@flink.apache.org>
Betreff: max aggregator dosen't work as expected

Hey,

I have some questions to aggregate functions such as max or min. Take the 
following example:

//create Stream with event time where Data contains an ID, a timestamp and a 
temperature value

DataStream<Data> oneStream = env.fromElements(

        new Data(123, new Date(116, 8,8,11,11,11), 5),

        new Data(124, new Date(116, 8,8,12,10,11), 8),

        new Data(125, new Date(116, 8,8,12,12,11), 10),

        new Data(126, new Date(116, 8,8,12,15,11), 2))

.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<Data>() {
    @Override
    public long extractAscendingTimestamp(final Data data) {
        return data.getTimestamp().getTime();
    }
});



//calluclate max value of temperature per hour

DataStream<Data> maxStream = WatermarkStream

        .keyBy("id")

        .timeWindow(Time.hours(1))

        .max("temp");
maxStream.print();




Here are the questions I ran into:

1.)    Why does the resulting stream “maxStream” have to be of type Data? From 
the documentation and the difference to maxBy I would expect the resulting 
stream to be of type Integer?

2.)    Executing the code as it is above, I would expect the printed result be 
the data with ID 124 and ID 125. However, the execution prints all 4 data sets. 
Did I totally get this example wrong? How would the code need  to look, to get 
the expected result?

3.)    Can you point me to a good example for using time windows and 
aggregates? I couldn’t find one that explains the above questions.

Thanks for your help and best wishes,
Claudia