Apache Flink Process Function state is not holding the state

Question

I am writing some code for a processElement function in Apache Flink 1.4:

public class ProcessFunctionClass extends ProcessFunction<Tuple2<String, String>, Tuple2<String, String>>{

    private ListState<String> listState;

    public void processElement(Tuple2<String, String> tuple2,  Context context, Collector<Tuple2<String, String>> collector) {

        // if the state is empty, start a timer
        if (listState.get().iterator().hasNext() == false)
            context.timerService().registerEventTimeTimer(10000);

        listState.add("someStringToBeStored");

        // ...
    }

}

I have this function for when the timer expires:

public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, String>> out) throws Exception {
    Iterable<String> strings = listState.get();
    int cnt = 0;
    int totalLength = 0;
    Iterator<String> it = strings.iterator();
    while (it.hasNext()) {
        cnt++;
        totalLength += it.next().length();
    }
    LOGGER.info("cnt is:" + cnt);
    LOGGER.info("totalLength is:" + totalLength);

    // clearing the state
    listState.clear();
}

However every time I run the application, the value of cnt is always 1, and the value of totalLength is the length of the particular string that has been processed at that time. It looks like the state is not kept in the system. From this code is it clear what I am doing wrong here?

David Anderson · Accepted Answer · 2020-06-21 06:13:41Z

2

Process functions used key-partitioned state, meaning there is a separate list for every key. My guess is that there is no key with multiple events in a 10 second period.

answered Jun 21, 2020 at 6:13

David Anderson

44.3k4 gold badges41 silver badges73 bronze badges

Sign up to request clarification or add additional context in comments.

Comments

kkrugler · Accepted Answer · 2020-06-21 00:50:46Z

0

Your ProcessFunctionClass needs to extend the Flink ProcessFunction.

answered Jun 21, 2020 at 0:50

kkrugler

9,5506 gold badges26 silver badges21 bronze badges

1 Comment

BlameMe Over a year ago

It does, I forgot to put it here. Edited.

Chandan Bansal · Accepted Answer · 2022-10-26 12:25:19Z

-1

The ProcessFunction is not meant to hold the state. You can use WindowProcessFunctions to hold the state of the elements present in the window.

answered Oct 26, 2022 at 12:25

Chandan Bansal

491 silver badge6 bronze badges

6 Comments

emilio Over a year ago

ProcessFunction can store a state. It requires to key the stream.

Chandan Bansal Over a year ago

@emilio do you have any working example where we can keep a global state in a process function, for eg. CoProcessFunction. Our goal being to access the current element when next element comes in.

emilio Over a year ago

Can you explain what you mean by global state? For my understanding flink does not maintain a "global" state. Have a look at nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/… And if you want to make it global you can key your entire stream by something like stream.keyBy(..."some static key").process(..you process fun)

Chandan Bansal Over a year ago

@emilio Lets assume we have a stream of kafka, the dtos from which we need to store in a mapstate. Now this mapstate currently only persists one entry. Meanining, if next entry comes in with key "xyz" we cannot check if mapstate already has a key named "xyz" because previous entry is gone now. That is what I mean by global state, i.e., one mapstate that should keep all the dtos of the stream.

emilio Over a year ago

As far as I understand you can use two keyBy. In the first one you use a static key keyBy("1") thus every incoming record will land on the same operator instance. After this you can key your stream again. But you should be aware that this can cause a bottleneck in your job and introduce further de/serialisation.

|

Collectives™ on Stack Overflow

Apache Flink Process Function state is not holding the state

3 Answers 3

Comments

1 Comment

6 Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

3 Answers 3

Comments

1 Comment

6 Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related