max aggregator dosen't work as expected

[Claudia Wegmann]

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

 

 

[Claudia Wegmann]

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 J

But question 1.) still remains.

 

Von: Claudia Wegmann [mailto:[hidden email]]
Gesendet: Montag, 8. August 2016 14:27
An: [hidden email]
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

 

 

 

[rmetzger0]

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 <[hidden email]> 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 J

But question 1.) still remains.

 

Von: Claudia Wegmann [mailto:[hidden email]]
Gesendet: Montag, 8. August 2016 14:27
An: [hidden email]
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

 

 

 

[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:[hidden email]]
Gesendet: Montag, 8. August 2016 15:52
An: [hidden email]
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 <[hidden email]> 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 J

But question 1.) still remains.

 

Von: Claudia Wegmann [mailto:[hidden email]]
Gesendet: Montag, 8. August 2016 14:27
An: [hidden email]
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