flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Claudia Wegmann <c.wegm...@kasasi.de>
Subject AW: max aggregator dosen't work as expected
Date Mon, 08 Aug 2016 14:12:27 GMT
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:rmetzger@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.wegmann@kasasi.de<mailto:c.wegmann@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.wegmann@kasasi.de<mailto:c.wegmann@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




Mime
View raw message