Re: How to sessionize stream with Apache Flink?

2017-06-19 Thread Fabian Hueske 
An alternative would be to use a FlatMapFunction with a ListState instead
of a window with custom trigger.

When a new element arrives (i.e., the flatMap() method is called), you
check if the value changed.
If the value did not changed, you append the element to the state.
If the value changed, you emit the current list state as a session, clear
the list, and insert the new element as the first to the list state.

However, you should keep in mind that this assumes that the order of
elements is preserved.
Flink ensures within a partition, i.e, as long as elements are not shuffled
and all operators run with the same parallelism.

Best, Fabian

2017-06-18 15:10 GMT+02:00 Jonas :

> Hey Milad,
>
> since you cannot look into the future which element comes next, you have to
> "lag" one behind. This requires building an operator that creates 2-tuples
> from incoming elements containing (current-1, current), so basically a
> single value state that emits the last and the current element in a tuple.
>
> In a trigger, the element is then of the 2-tuple type and you can see
> changes "beforehand". The last element of 1's is then (1, 2).
>
> Hope this helps.
>
>
>
> --
> View this message in context: http://apache-flink-user-
> mailing-list-archive.2336050.n4.nabble.com/How-to-
> sessionize-stream-with-Apache-Flink-tp13817p13818.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive
> at Nabble.com.
>

Re: How to sessionize stream with Apache Flink?

2017-06-18 Thread Jonas 
Hey Milad,

since you cannot look into the future which element comes next, you have to
"lag" one behind. This requires building an operator that creates 2-tuples
from incoming elements containing (current-1, current), so basically a
single value state that emits the last and the current element in a tuple.

In a trigger, the element is then of the 2-tuple type and you can see
changes "beforehand". The last element of 1's is then (1, 2).

Hope this helps.



--
View this message in context: 
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/How-to-sessionize-stream-with-Apache-Flink-tp13817p13818.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at 
Nabble.com.

How to sessionize stream with Apache Flink?

2017-06-18 Thread Milad khajavi 
Hi guys,
I want to sessionize this stream: 1,1,1,2,2,2,2,2,3,3,3,3,3,3,3,0,3,3,3,5,
... to these sessions:

1,1,1
2,2,2,2,2
3,3,3,3,3,3,3
0
3,3,3
5

I've wrote CustomTrigger to detect when stream elements change from 1 to 2
(2 to 3, 3 to 0 and so on) and then fire the trigger. But this is not the
solution, because when I processing the first element of 2's, and fire the
trigger the window will be [1,1,1,2] but I need to fire the trigger on the
last element of 1's.


Here is the pesudo of my onElement function in my custom trigger class:

override def onElement(element: Session, timestamp: Long, window: W, ctx:
TriggerContext): TriggerResult = {
if (prevState == element.value) {
  prevState = element.value
  TriggerResult.CONTINUE
} else {
  prevState = element.value
  TriggerResult.FIRE
}
}

How can I solve this problem?

-- 
Milād Khājavi
http://blog.khajavi.ir
Having the source means you can do it yourself.
I tried to change the world, but I couldn’t find the source code.