Hello Aljoscha <aljos...@apache.org>,
My sincere apologies at the beginning, if I seem to repeat the same
question, almost interminably. If it is frustrating you, I seek your
patience but I really want to nail it down in mind. :-)
The point about parallelizing is well taken. I understand why the stream
should be broken into multiple partitions and how. The understanding that
is still evading me is how is the use-case of computing an (sliding)
average temperature achieved if the stream is scattered.
I want the *running* average temperature for every 3 readings, sliding by 1
reading. I am monitoring the average temperature; if it goes beyond a
certain threshold for 3 consecutive readings, I throw an alarm.
Let's take the following set of data (fields are: probeID, timestamp,
temperature ; 'timestamp' field is used for assignAscendingTimestamp()
function):
P1,T1,20
P1,T2,30
P2,T2,30
P1,T3,50
P2,T3,20
P3,T3,10
Assumption: T1 < T2 < T3
Now, if we partition on the probeID, we get three partitions, thus:
P1 -> (T1,20) | (T2,30) | (T3,50)
P2 -> (T2,30) | (T3,20)
P3 -> (T3,10)
Computing the average temperature will give me *three distinct averages*
here, one for each partition. I get average per probe, not per every 3
readings [assuming a slidingWindow(3,1)] irrespective of which probe gives.
Is it even correct to expect a running average if we partition the input
stream?
Hope I am making my understanding (or the lack of it), quite clear here! :-)
-- Nirmalya
-------------------------------------------------------------------------------------------------------------------------------------------------------------
To: user@flink.apache.org
Cc:
Date: Fri, 19 Feb 2016 10:41:52 +0100
Subject:
Hi,
as I understand it the “temp_reading_timestamp” field is not a key on which
you can partition your data. This is a field that would be used for
assigning the elements to timestamps. In you data you also have the
“probeID” field. This is a field that could be used to parallelize
computation, for example you could
do the following:
val inputStream = <define some source>
val result = inputStream
.assignAscendingTimestamps { e => e.temp_reading_timestamp }
.keyBy { e => e.probeID }
.timeWindow(Time.minutes(10))
.apply(new SumFunction(), new ComputeAverageFunction())
result.print()
(Where SumFunction() would sum up temperatures and keep a count and
ComputeAverageFunction() would divide the sum by the count.)
In this way, computation is parallelized because it can be spread across
several machines and partitioned by the key. Without such a key everything
has to be computed on one machine because a global view of the data is
required.
Cheers,
--
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."
