flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Stefano Baghino <stefano.bagh...@radicalbit.io>
Subject Re: Finding the average temperature
Date Sun, 14 Feb 2016 19:22:58 GMT
Hello again Nirmalaya,

in this case you are keying by timestamp; think of keying as grouping: this
means that windows are brought together according to their timestamp. I
misread your original post but now that I see the code I understand your
problem.

I've put some code here:
https://github.com/stefanobaghino/flink-stream-avg-example

It should more or less be the same as your program. What I did was removing
the keying. By not keying the stream, window operations (like
countWindowAll instead of countWindow) act on the whole stream, thus
forcing a parallelism of 1. I've put a couple of prints so that you can
visualize what is the parallelism and what goes in each window. To get a
better understanding of what was going on, you can put the same prints in
your code as well (printing the key for each keyed window as well); you'll
see that in your case windows are "grouped" by timestamp.

Hope I've been a little bit more helpful than last time. :)

On Sun, Feb 14, 2016 at 6:50 PM, Nirmalya Sengupta <
sengupta.nirmalya@gmail.com> wrote:

> Hello Stefano <stefano.baghino@radicalbit.io>
>
> I have tried to implement what I understood from your mail earlier in the
> day, but it doesn't seem to produce the result I expect. Here's the code
> snippet:
>
> -------------------------------------------------------------------------
>     val env = StreamExecutionEnvironment.createLocalEnvironment(4)
>
>     env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
>
>     val readings =
>       readIncomingReadings(env,"./sampleIOTTiny.csv")
>         .keyBy(_.readingTimeStamp)
>         .countWindow(4,2)
>
>     val avgReadings =
>         readings
>         .apply((key: Long,w: Window,v: Iterable[IncomingDataUnit],out:
> Collector[Float]) => {
>
>           val readings : Iterable[Float] = v.map(_.ambientTemperature)
>           val avg = readings.sum / readings.size
>
>           out.collect(avg)
>
>     }).setParallelism(1)
>
>     avgReadings.print()
>
> -------------------------------------------------------------------------
>
> And, here's the output:
>
> -------------------------------------------------------------------------
> 1> 23.67
> 1> 21.0025
> 1> 23.79
> 2> 25.02
> 2> 23.3425
> 2> 25.02
> 3> 26.55
> 4> 19.970001
> 3> 18.93375
> 4> 25.727499
> 3> 18.93375
> 4> 25.7075
> --------------------------------------------------------------------------
>
> My understanding is that because I have associated a parallelism(1) to the
> avgReadings transformation, it should aggregate the streams from all the 4
> earlier windows, and then compute the single average value. It is quite
> apparent that there is a gap in my understanding. Could you please point
> out the mistake that I am making?
>
> Many thanks in advance.
>
> -- Nirmalya
>
> --
> Software Technologist
> http://www.linkedin.com/in/nirmalyasengupta
> "If you have built castles in the air, your work need not be lost. That is
> where they should be.
> Now put the foundation under them."
>



-- 
BR,
Stefano Baghino

Software Engineer @ Radicalbit

Mime
View raw message