Re: periodic trigger

[Piotr Nowojski]

Hi,

Sorry for late response (because of the holiday period).

You didn’t mention lateness previously, that’s why I proposed such solution. 

Another approach would be to calculate max session length per user on the first 
aggregation level and at the same time remember what was the previously 
emitted/triggered value. Now, whenever you recalculate your window because of 
firing a trigger, you could check what is the new value and what was the 
previously emitted value. If they are the same, you do not have to emit 
anything. If they are different, you would have to issue an “update”/“diff” to 
the global aggregation on the second level. In case of simple average of 
session length, first level on update could emit “-old_value” and “+new_value” 
(negative old_value and positive new_value). 

To do this in Flink you could use 

org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction

and store the previously emitted values in

org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction.Context#windowState()

For efficiency reasons best to combine it with a reduce function using this call

org.apache.flink.streaming.api.datastream.WindowedStream#reduce(org.apache.flink.api.common.functions.ReduceFunction,
 
org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction)

Reduce will ensure that your ProcessWindowFunction will not have to process all 
events belonging to the window each time it is triggered, but only it will have 
to process the single reduced element.

In your case:

.keyBy("userId", "sessionId")
.window(TumblingEventTimeWindows.of(Time.seconds(60)))
.trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(10)))
.reduce(MAX_BY_LENGTH_REDUCE_FUNCTION, 
MY_FANCY_PROCESS_WINDOW_FUNCTION_WITH_UPDATES).

Of course second level global aggregation would have to understand and take 
those “update”/“diffs” into account, but that would be simple to implement by 
some custom reduce function/aggregate. You would like the sequence of values 1, 
1, 10, -10, 1 produce average of 1 and not 0.6 - negative values have to 
decrease counters in calculating the average value (to aggregate average value 
you always need to keep sum of values and a counter).

.timeWindowAll(Time.seconds(60))
.aggregate(FANCY_AGGREGATE_THAT_HANDLES_UPDATES)
.print();

Hope that helps.

Piotrek

> On 22 Dec 2017, at 14:10, Plamen Paskov  wrote:
> 
> I think it will not solve the problem as if i set ContinuousEventTimeTrigger 
> to 10 seconds and allowedLateness(Time.seconds(60)) as i don't want to 
> discard events from different users received later then i might receive more 
> than one row for a single user based on the number of windows created by the 
> events of this user. That will make the the average computations wrong.
> 
> On 22.12.2017 12:10, Piotr Nowojski wrote:
>> Ok, I think now I understand your problem. 
>> 
>> Wouldn’t it be enough, if you change last global window to something like 
>> this:
>> 
>> lastUserSession
>> .timeWindowAll(Time.seconds(10))
>> .aggregate(new AverageSessionLengthAcrossAllUsers())
>> .print();
>> 
>> (As a side note, maybe you should use ContinousEventTimeTrigger in the first 
>> window). This way it will aggregate and calculate average session length of 
>> only last “preview results” of the 60 seconds user windows (emitted every 10 
>> seconds from the first aggregation).
>> 
>> Piotrek
>> 
>>> On 21 Dec 2017, at 15:18, Plamen Paskov >> > wrote:
>>> 
>>> Imagine a case where i want to run a computation every X seconds for 1 day 
>>> window. I want the calculate average session length for current day every X 
>>> seconds. Is there an easy way to achieve that?
>>> 
>>> On 21.12.2017 16:06, Piotr Nowojski wrote:
 Hi,
 
 You defined a tumbling window 
 (https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/windows.html#tumbling-windows
  
 )
  of 60 seconds, triggered every 10 seconds. This means that each input 
 element can be processed/averaged up to 6 times (there is no other way if 
 you trigger each window multiple times).
 
 I am not sure what are you trying to achieve, but please refer to the 
 documentation about different window types (tumbling, sliding, session) 
 maybe it will clarify things for you:
 https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/windows.html#window-assigners
  
 
 
 If you want to avoid duplicated processing, use either tumbling window 
 with default trigger (triggering at the end of the window), or use session 
 windows.
 
 Piotrek

Re: periodic trigger

[Plamen Paskov]

I think it will not solve the problem as if i set 
ContinuousEventTimeTrigger to 10 seconds and 
allowedLateness(Time.seconds(60)) as i don't want to discard events from 
different users received later then i might receive more than one row 
for a single user based on the number of windows created by the events 
of this user. That will make the the average computations wrong.



On 22.12.2017 12:10, Piotr Nowojski wrote:

Ok, I think now I understand your problem.

Wouldn’t it be enough, if you change last global window to something 
like this:


lastUserSession
 .timeWindowAll(*Time.seconds(10)*)
 .aggregate(new AverageSessionLengthAcrossAllUsers())
 .print();

(As a side note, maybe you should use ContinousEventTimeTrigger in the 
first window). This way it will aggregate and calculate average 
session length of only last “preview results” of the 60 seconds user 
windows (emitted every 10 seconds from the first aggregation).


Piotrek

On 21 Dec 2017, at 15:18, Plamen Paskov 
> wrote:


Imagine a case where i want to run a computation every X seconds for 
1 day window. I want the calculate average session length for current 
day every X seconds. Is there an easy way to achieve that?



On 21.12.2017 16:06, Piotr Nowojski wrote:

Hi,

You defined a tumbling window 
(https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/windows.html#tumbling-windows) 
of 60 seconds, triggered every 10 seconds. This means that each 
input element can be processed/averaged up to 6 times (there is no 
other way if you trigger each window multiple times).


I am not sure what are you trying to achieve, but please refer to 
the documentation about different window types (tumbling, sliding, 
session) maybe it will clarify things for you:

https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/windows.html#window-assigners

If you want to avoid duplicated processing, use either tumbling 
window with default trigger (triggering at the end of the window), 
or use session windows.


Piotrek


On 21 Dec 2017, at 13:29, Plamen Paskov 
> wrote:


Hi guys,
I have the following code:

SingleOutputStreamOperator lastUserSession = env
 .socketTextStream("localhost",9000,"\n")
 .map(new MapFunction() {
 @Override public Event map(String value)throws Exception {
 String[] row = value.split(",");
 return new Event(Long.valueOf(row[0]), row[1], 
Long.valueOf(row[2]), Timestamp.valueOf(row[3]).getTime());
 }
 })
 .assignTimestampsAndWatermarks(new 
BoundedOutOfOrdernessTimestampExtractor(Time.seconds(10)) {
 @Override public long extractTimestamp(Event element) {
 return element.timestamp;
 }
 })
 .keyBy("userId","sessionId")
 .window(TumblingEventTimeWindows.of(Time.seconds(60)))
 .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(10)))
 .maxBy("length",false);

lastUserSession
 .timeWindowAll(Time.seconds(60))
 .aggregate(new AverageSessionLengthAcrossAllUsers())
 .print();

What i'm trying to achieve is to calculate the average session length every 10 
seconds. The problem is that once the window length is 60 seconds and a 
computation is triggered
every 10 seconds i will receive duplicate events in my average calculation 
method so the average will not be correct. If i move 
ContinuousProcessingTimeTrigger down before
AverageSessionLengthAcrossAllUsers() then it's not triggering every 10 seconds.
Any other suggestions how to workaround this?

Thanks

Re: periodic trigger

[Piotr Nowojski]

Ok, I think now I understand your problem. 

Wouldn’t it be enough, if you change last global window to something like this:

lastUserSession
.timeWindowAll(Time.seconds(10))
.aggregate(new AverageSessionLengthAcrossAllUsers())
.print();

(As a side note, maybe you should use ContinousEventTimeTrigger in the first 
window). This way it will aggregate and calculate average session length of 
only last “preview results” of the 60 seconds user windows (emitted every 10 
seconds from the first aggregation).

Piotrek

> On 21 Dec 2017, at 15:18, Plamen Paskov  wrote:
> 
> Imagine a case where i want to run a computation every X seconds for 1 day 
> window. I want the calculate average session length for current day every X 
> seconds. Is there an easy way to achieve that?
> 
> On 21.12.2017 16:06, Piotr Nowojski wrote:
>> Hi,
>> 
>> You defined a tumbling window 
>> (https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/windows.html#tumbling-windows
>>  
>> )
>>  of 60 seconds, triggered every 10 seconds. This means that each input 
>> element can be processed/averaged up to 6 times (there is no other way if 
>> you trigger each window multiple times).
>> 
>> I am not sure what are you trying to achieve, but please refer to the 
>> documentation about different window types (tumbling, sliding, session) 
>> maybe it will clarify things for you:
>> https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/windows.html#window-assigners
>>  
>> 
>> 
>> If you want to avoid duplicated processing, use either tumbling window with 
>> default trigger (triggering at the end of the window), or use session 
>> windows.
>> 
>> Piotrek
>> 
>> 
>>> On 21 Dec 2017, at 13:29, Plamen Paskov >> > wrote:
>>> 
>>> Hi guys,
>>> I have the following code:
>>> 
>>> SingleOutputStreamOperator lastUserSession = env
>>> .socketTextStream("localhost", 9000, "\n")
>>> .map(new MapFunction() {
>>> @Override
>>> public Event map(String value) throws Exception {
>>> String[] row = value.split(",");
>>> return new Event(Long.valueOf(row[0]), row[1], 
>>> Long.valueOf(row[2]), Timestamp.valueOf(row[3]).getTime());
>>> }
>>> })
>>> .assignTimestampsAndWatermarks(new 
>>> BoundedOutOfOrdernessTimestampExtractor(Time.seconds(10)) {
>>> @Override
>>> public long extractTimestamp(Event element) {
>>> return element.timestamp;
>>> }
>>> })
>>> .keyBy("userId", "sessionId")
>>> .window(TumblingEventTimeWindows.of(Time.seconds(60)))
>>> .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(10)))
>>> .maxBy("length", false);
>>> 
>>> lastUserSession
>>> .timeWindowAll(Time.seconds(60))
>>> .aggregate(new AverageSessionLengthAcrossAllUsers())
>>> .print();
>>> 
>>> What i'm trying to achieve is to calculate the average session length every 
>>> 10 seconds. The problem is that once the window length is 60 seconds and a 
>>> computation is triggered
>>> every 10 seconds i will receive duplicate events in my average calculation 
>>> method so the average will not be correct. If i move 
>>> ContinuousProcessingTimeTrigger down before 
>>> AverageSessionLengthAcrossAllUsers() then it's not triggering every 10 
>>> seconds.
>>> Any other suggestions how to workaround this?
>>> 
>>> Thanks
>> 
> 

Re: periodic trigger

[Plamen Paskov]

Imagine a case where i want to run a computation every X seconds for 1 
day window. I want the calculate average session length for current day 
every X seconds. Is there an easy way to achieve that?



On 21.12.2017 16:06, Piotr Nowojski wrote:

Hi,

You defined a tumbling window 
(https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/windows.html#tumbling-windows) 
of 60 seconds, triggered every 10 seconds. This means that each input 
element can be processed/averaged up to 6 times (there is no other way 
if you trigger each window multiple times).


I am not sure what are you trying to achieve, but please refer to the 
documentation about different window types (tumbling, sliding, 
session) maybe it will clarify things for you:

https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/windows.html#window-assigners

If you want to avoid duplicated processing, use either tumbling window 
with default trigger (triggering at the end of the window), or use 
session windows.


Piotrek


On 21 Dec 2017, at 13:29, Plamen Paskov 
> wrote:


Hi guys,
I have the following code:

SingleOutputStreamOperator lastUserSession = env
 .socketTextStream("localhost",9000,"\n")
 .map(new MapFunction() {
 @Override public Event map(String value)throws Exception {
 String[] row = value.split(",");
 return new Event(Long.valueOf(row[0]), row[1], 
Long.valueOf(row[2]), Timestamp.valueOf(row[3]).getTime());
 }
 })
 .assignTimestampsAndWatermarks(new 
BoundedOutOfOrdernessTimestampExtractor(Time.seconds(10)) {
 @Override public long extractTimestamp(Event element) {
 return element.timestamp;
 }
 })
 .keyBy("userId","sessionId")
 .window(TumblingEventTimeWindows.of(Time.seconds(60)))
 .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(10)))
 .maxBy("length",false);

lastUserSession
 .timeWindowAll(Time.seconds(60))
 .aggregate(new AverageSessionLengthAcrossAllUsers())
 .print();

What i'm trying to achieve is to calculate the average session length every 10 
seconds. The problem is that once the window length is 60 seconds and a 
computation is triggered
every 10 seconds i will receive duplicate events in my average calculation 
method so the average will not be correct. If i move 
ContinuousProcessingTimeTrigger down before
AverageSessionLengthAcrossAllUsers() then it's not triggering every 10 seconds.
Any other suggestions how to workaround this?

Thanks

Re: periodic trigger

[Piotr Nowojski]

Hi,

You defined a tumbling window 
(https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/windows.html#tumbling-windows
 
)
 of 60 seconds, triggered every 10 seconds. This means that each input element 
can be processed/averaged up to 6 times (there is no other way if you trigger 
each window multiple times).

I am not sure what are you trying to achieve, but please refer to the 
documentation about different window types (tumbling, sliding, session) maybe 
it will clarify things for you:
https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/windows.html#window-assigners
 


If you want to avoid duplicated processing, use either tumbling window with 
default trigger (triggering at the end of the window), or use session windows.

Piotrek


> On 21 Dec 2017, at 13:29, Plamen Paskov  wrote:
> 
> Hi guys,
> I have the following code:
> 
> SingleOutputStreamOperator lastUserSession = env
> .socketTextStream("localhost", 9000, "\n")
> .map(new MapFunction() {
> @Override
> public Event map(String value) throws Exception {
> String[] row = value.split(",");
> return new Event(Long.valueOf(row[0]), row[1], 
> Long.valueOf(row[2]), Timestamp.valueOf(row[3]).getTime());
> }
> })
> .assignTimestampsAndWatermarks(new 
> BoundedOutOfOrdernessTimestampExtractor(Time.seconds(10)) {
> @Override
> public long extractTimestamp(Event element) {
> return element.timestamp;
> }
> })
> .keyBy("userId", "sessionId")
> .window(TumblingEventTimeWindows.of(Time.seconds(60)))
> .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(10)))
> .maxBy("length", false);
> 
> lastUserSession
> .timeWindowAll(Time.seconds(60))
> .aggregate(new AverageSessionLengthAcrossAllUsers())
> .print();
> 
> What i'm trying to achieve is to calculate the average session length every 
> 10 seconds. The problem is that once the window length is 60 seconds and a 
> computation is triggered
> every 10 seconds i will receive duplicate events in my average calculation 
> method so the average will not be correct. If i move 
> ContinuousProcessingTimeTrigger down before 
> AverageSessionLengthAcrossAllUsers() then it's not triggering every 10 
> seconds.
> Any other suggestions how to workaround this?
> 
> Thanks