Improving Ignite window processing support

[Roman Shtykh]

Igniters,
I was thinking about improving Ignite window processing support so that queries for windowed data is not user-initiated (which can have timing issues) but rather event-driven, similar to ContinuousQuery but being able to fire size-based or time-based entry windows to the listening client.
Some thougths on implementation:1. Guarantee all window events go to the same partition (user will need to specify it as a windowed cache). Maybe it can be rather implemented by extending IgniteQueue?2. Set a trigger on cache, which will listen to eviction (in case of size-based windowed cache) or expiration (time-based cache) events of the cache and fire entries.
It has to be exposed to the user by some API, very roughly something likeIgniteStream<K,V> is = IgniteStream.on(cache).with(windowingType, filterPredidate).aggregate(aggFunc)is.run() // to continuously return windowed results
where windowingType is time/size/session-based windows, aggFunc is sum/min/max/etc. or user-specified.
It can be an experimental feature and I think it is a useful API to enforce our stream processing, but I would like to know your opinion. Do you think such API is needed? I have a limited knowledge of Ignite internals -- any other ideas on the implementation?
For your reference, https://flink.apache.org/news/2015/12/04/Introducing-windows.html is a good introduction on windows processing.
-Roman

P.S. Other things like operator/event/ingress time but has to be considered for the implementation are omitted.

[Murthy Kakarlamudi]

Thanks for thinking about this feature. Due to lack of this feature, in our
application we are currently planning to have some Timer based classes
query the cache, perform aggregations and send the events to all the
listeners.
Having this feature built in into Ignite certainly helps.

Satya.
On Mar 25, 2016 10:38 PM, "Roman Shtykh" <[hidden email]> wrote:

> Igniters,
> I was thinking about improving Ignite window processing support so that
> queries for windowed data is not user-initiated (which can have timing
> issues) but rather event-driven, similar to ContinuousQuery but being able
> to fire size-based or time-based entry windows to the listening client.
> Some thougths on implementation:1. Guarantee all window events go to the
> same partition (user will need to specify it as a windowed cache). Maybe it
> can be rather implemented by extending IgniteQueue?2. Set a trigger on
> cache, which will listen to eviction (in case of size-based windowed cache)
> or expiration (time-based cache) events of the cache and fire entries.
> It has to be exposed to the user by some API, very roughly something
> likeIgniteStream<K,V> is = IgniteStream.on(cache).with(windowingType,
> filterPredidate).aggregate(aggFunc)is.run() // to continuously return
> windowed results
> where windowingType is time/size/session-based windows, aggFunc is
> sum/min/max/etc. or user-specified.
> It can be an experimental feature and I think it is a useful API to
> enforce our stream processing, but I would like to know your opinion. Do
> you think such API is needed? I have a limited knowledge of Ignite
> internals -- any other ideas on the implementation?
> For your reference,
> https://flink.apache.org/news/2015/12/04/Introducing-windows.html is a
> good introduction on windows processing.
> -Roman
>
> P.S. Other things like operator/event/ingress time but has to be
> considered for the implementation are omitted.
>

[dsetrakyan]

Roman,

This is an excellent idea!

The API you are suggesting is very close to the reactive streams APIs [1],
so my suggestion would be to simply implement that API.

Another thing to focus on would be to provide guarantees:
- at least once
- at most once
- exactly once

I am not suggesting that we do it all in one step, but it would help to
identify what will be supported in the first iteration.

Do you mind taking a first step at this? We can discuss all questions
either here or in the chat room.

[1] http://www.reactive-streams.org/

D.

On Fri, Mar 25, 2016 at 8:08 PM, Murthy Kakarlamudi <[hidden email]>
wrote:

> Thanks for thinking about this feature. Due to lack of this feature, in our
> application we are currently planning to have some Timer based classes
> query the cache, perform aggregations and send the events to all the
> listeners.
> Having this feature built in into Ignite certainly helps.
>
> Satya.
> On Mar 25, 2016 10:38 PM, "Roman Shtykh" <[hidden email]>
> wrote:
>
> > Igniters,
> > I was thinking about improving Ignite window processing support so that
> > queries for windowed data is not user-initiated (which can have timing
> > issues) but rather event-driven, similar to ContinuousQuery but being
> able
> > to fire size-based or time-based entry windows to the listening client.
> > Some thougths on implementation:1. Guarantee all window events go to the
> > same partition (user will need to specify it as a windowed cache). Maybe
> it
> > can be rather implemented by extending IgniteQueue?2. Set a trigger on
> > cache, which will listen to eviction (in case of size-based windowed
> cache)
> > or expiration (time-based cache) events of the cache and fire entries.
> > It has to be exposed to the user by some API, very roughly something
> > likeIgniteStream<K,V> is = IgniteStream.on(cache).with(windowingType,
> > filterPredidate).aggregate(aggFunc)is.run() // to continuously return
> > windowed results
> > where windowingType is time/size/session-based windows, aggFunc is
> > sum/min/max/etc. or user-specified.
> > It can be an experimental feature and I think it is a useful API to
> > enforce our stream processing, but I would like to know your opinion. Do
> > you think such API is needed? I have a limited knowledge of Ignite
> > internals -- any other ideas on the implementation?
> > For your reference,
> > https://flink.apache.org/news/2015/12/04/Introducing-windows.html is a
> > good introduction on windows processing.
> > -Roman
> >
> > P.S. Other things like operator/event/ingress time but has to be
> > considered for the implementation are omitted.
> >
>

[Roman Shtykh]

Dmitriy,
Honestly I was thinking only of window processing to cover only this specific area. But I think looking at reactive streams makes sense and window processing can be implemented within reactive streams with all three delivery guarantee semantics (starting with the easiest two).
Also I am interested in working on it. I will create a JIRA ticket for this and may have many questions ;)
-Roman
 

    On Sunday, March 27, 2016 5:03 AM, Dmitriy Setrakyan <[hidden email]> wrote:
 

 Roman,

This is an excellent idea!

The API you are suggesting is very close to the reactive streams APIs [1],
so my suggestion would be to simply implement that API.

Another thing to focus on would be to provide guarantees:
- at least once
- at most once
- exactly once

I am not suggesting that we do it all in one step, but it would help to
identify what will be supported in the first iteration.

Do you mind taking a first step at this? We can discuss all questions
either here or in the chat room.

[1] http://www.reactive-streams.org/

D.

On Fri, Mar 25, 2016 at 8:08 PM, Murthy Kakarlamudi <[hidden email]>
wrote:

> Thanks for thinking about this feature. Due to lack of this feature, in our
> application we are currently planning to have some Timer based classes
> query the cache, perform aggregations and send the events to all the
> listeners.
> Having this feature built in into Ignite certainly helps.
>
> Satya.
> On Mar 25, 2016 10:38 PM, "Roman Shtykh" <[hidden email]>
> wrote:
>
> > Igniters,
> > I was thinking about improving Ignite window processing support so that
> > queries for windowed data is not user-initiated (which can have timing
> > issues) but rather event-driven, similar to ContinuousQuery but being
> able
> > to fire size-based or time-based entry windows to the listening client.
> > Some thougths on implementation:1. Guarantee all window events go to the
> > same partition (user will need to specify it as a windowed cache). Maybe
> it
> > can be rather implemented by extending IgniteQueue?2. Set a trigger on
> > cache, which will listen to eviction (in case of size-based windowed
> cache)
> > or expiration (time-based cache) events of the cache and fire entries.
> > It has to be exposed to the user by some API, very roughly something
> > likeIgniteStream<K,V> is = IgniteStream.on(cache).with(windowingType,
> > filterPredidate).aggregate(aggFunc)is.run() // to continuously return
> > windowed results
> > where windowingType is time/size/session-based windows, aggFunc is
> > sum/min/max/etc. or user-specified.
> > It can be an experimental feature and I think it is a useful API to
> > enforce our stream processing, but I would like to know your opinion. Do
> > you think such API is needed? I have a limited knowledge of Ignite
> > internals -- any other ideas on the implementation?
> > For your reference,
> > https://flink.apache.org/news/2015/12/04/Introducing-windows.html is a
> > good introduction on windows processing.
> > -Roman
> >
> > P.S. Other things like operator/event/ingress time but has to be
> > considered for the implementation are omitted.
> >
>