AW: max aggregator dosen't work as expected
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
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 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() {
@Override
public long extractAscendingTimestamp(final Data data) {
return data.getTimestamp().getTime();
}
});
//calluclate max value of temperature per hour
DataStream 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
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 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:c.wegm...@kasasi.de]
> *Gesendet:* Montag, 8. August 2016 14:27
> *An:* 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 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() {
> @Override
> *public long *extractAscendingTimestamp(*final *Data data) {
> *return *data.getTimestamp().getTime();
> }
> });
>
>
>
> //calluclate max value of temperature per hour
>
> DataStream 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
>
>
>
>
>
>
>
AW: max aggregator dosen't work as expected
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]
Gesendet: Montag, 8. August 2016 14:27
An: 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 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() {
@Override
public long extractAscendingTimestamp(final Data data) {
return data.getTimestamp().getTime();
}
});
//calluclate max value of temperature per hour
DataStream 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
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 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() {
@Override
public long extractAscendingTimestamp(final Data data) {
return data.getTimestamp().getTime();
}
});
//calluclate max value of temperature per hour
DataStream 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
