I would like to run an operation within Flink which yields a specific result. For computing the result I need to access the last <n> inputs of this operation. How would I do that? What do I have to take care with regards to checkpointing or watermarking?

Not sure I fully understand, but it sounds a bit like what you are looking for is to key by some operation_id and then use a count window to group a certain number of records per key to perform some calculation. Does that sound right?

Maybe. 🙂 Lets have some values: v0 v1 v2 v3 v4 v5 ... In this examples the calculation requires 3 values. So calc(), calc(v0) and calc(v0,v1) would yield an invalid output. But starting with calc(v0,v1,v2) we get a valid output, the next would be calc(v1,v2,v3) and so on. In the real use case we would instead of 3 input we would have several dozens and the values would be more like a data structure then a single value.

You can use state to persist the last N values: https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/fault-tolerance/state/#using-keyed-state

Hi Maciej, thanks for the link. I assume I use the ListState? I saw that I can call

add

to add the newest value. But there seems to be no

remove

to remove the oldest. That means I would need to call

clear

and

addAll

. Would that be correct? How would that affect the performance?

You can do what you say, or use something like

ValueState<DataStructure<T>>

- I think the only problem with that is that you always have to serialize and deserialize whole structure when you refer to it. At least if you use the RocksDB state backend.

One approach to avoiding large state deserialization/re-serialization is to use

MapState<Integer, YourValueType>

, where the key is an incrementing value. Then you have two

ValueState<Integer>

states, one for the lowest index, and the other for the highest index. When you add a value, you’d increment the highest index, then use the two index states to determine the number of records. If you need to flush, you can remove the specific element(s) from the

MapState

, then update your lowest index state. RocksDB stores each row as a separate object, thus you avoid de/re-serialization of all state.

Thanks for all the ideas. I'll look into them.