Keep keyed state across multiple transformations

Question

I have a stream that I want to partition using a certain key, and then run through several transformations, each using a state. When I call keyBy() I get a KeyedStream and the next transformation can access a partitioned state correctly, but another transformation chained after that gets an exception when trying to access a partitioned state. The exception is:

State key serializer has not been configured in the config. This operation cannot use partitioned state

It seems that the key information is only passed to the first transformation and not further down the chain.

The code I try to run is along the lines of this code (but actually does something):

DataStream<Long> newStream = eventsStream
    .keyBy("username")
    .filter(new RichFilterFunction<Event>() {
        private ValueState<Boolean> stateStore;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            stateStore = getRuntimeContext().getState(new ValueStateDescriptor<>("VALUE1", Boolean.class, Boolean.TRUE));
        }

        @Override
        public boolean filter(Event value) throws Exception {
            return stateStore.value();
        }
    })
    .map(new RichMapFunction<Event, Long>() {
        private ValueState<Long> stateStore;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            stateStore = getRuntimeContext().getState(new ValueStateDescriptor<>("VALUE2", Long.class, 0L));
        }

        @Override
        public Long map(Event value) throws Exception {
            return Long.parseLong(value.data) + stateStore.value();
        }
    });

This code will throw an exception at the second getState() call.

I can call keyBy() again, but then I remove the ability to chain the operations. Can I manually manipulate the objects of the stream graph so that the key information is passed, or is this sort of chaining not supported?

Matthias J. Sax · Accepted Answer · 2016-08-16T16:18:04.870

3

You can not.

Even if you would call keyBy() a second time (or somehow pass the "key-ed" information downstream), you would get a new state because a state is associated to a single operator only.

As as workaround, you need to merge both operators into one.

If you think this feature might be helpful, feel free to suggest it at dev@flink.apache.org.

edited Aug 16 '16 at 16:18

answered Aug 16 '16 at 09:29

Matthias J. Sax

59,682
7
117
137

1

Thank you, but it was an example. The operations can also be reduce() and map(). In theory you can also combine them into a flatMap(), but it makes more sense to keep them as individual operations. – Avihai Berkovitz Aug 16 '16 at 14:44
Extended my answer. – Matthias J. Sax Aug 16 '16 at 16:18
Thanks, but It is a basic feature of data streaming services - chain an operations. Why would suggest feature? – Albert Bikeev Apr 04 '18 at 15:49
Operator chaining is a physical operation, not a logical one. Thus, even if both are chained, logically the second operator does not have access to the state. – Matthias J. Sax Apr 04 '18 at 20:27

Keep keyed state across multiple transformations

1 Answers1

Linked