Lookup from state

[Taher Koitawala]

Hi All,

         As per my knowledge, all windowing operators in flink are stateful. So let's say I have 2 streams, Stream1 and Stream2.

Stream1 and Stream2 are aggregated over some key and which is then windowed on EventTime. So record X from Stream1 reaches flink on time, however, record X' from Stream2 reaches after maxOutOfOrderness time has passed. In this scenario as per my knowledge. X will be maintained in the flink state. However, when X' comes, how do I do a lookup for X from the flink state and carry on the further aggregation or whatever I want to do.

[Kostas Kloudas]

Hi Taher,

To understand your use case, you have something like the following:

stream1.keyBy(…)

.connect(stream2.keyBy(…))

.window(…).apply(MyWindowFunction)

and you want from within the MyWindowFunction to access the state for a FIRED window when a late element arrives for that key and window?

In this case, it depends what you want to do. 

First of all, with allowed lateness = 0, after the outofOrderness expires, then the state is no longer there, so you cannot access it.

If you set the allowedLateness to > 0, then the state will be there until the allowed lateness expires (see [1]), but in order to help with your use case,

it would help if you shared a bit more details about it. The reason is that form within the MyWindowFunction, and depending on if 

you use apply() or reduce() or aggregate() or if you use an evictor, there are specific things that you can do, and things that you cannot do.

Finally, for more control you may need to have a look at the CoProcessFunction in [2].

I hope this helps,

Kostas

[1] https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/operators/windows.html

[2] https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/operators/process_function.html

On Sep 17, 2018, at 3:15 PM, Taher Koitawala <[hidden email]> wrote:

Hi All,

         As per my knowledge, all windowing operators in flink are stateful. So let's say I have 2 streams, Stream1 and Stream2.

Stream1 and Stream2 are aggregated over some key and which is then windowed on EventTime. So record X from Stream1 reaches flink on time, however, record X' from Stream2 reaches after maxOutOfOrderness time has passed. In this scenario as per my knowledge. X will be maintained in the flink state. However, when X' comes, how do I do a lookup for X from the flink state and carry on the further aggregation or whatever I want to do.