Apache Flink Process Function 状态未保持状态答案

【问题标题】：Apache Flink Process Function state is not holding the stateApache Flink Process Function 状态未保持状态
【发布时间】：2020-06-20 22:38:30
【问题描述】：

我正在为 Apache Flink 1.4 中的 processElement 函数编写一些代码：

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");

        // ...
    }

}

当定时器到期时，我有这个功能：

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();
}

但是每次我运行应用程序时，cnt 的值始终为 1，totalLength 的值是当时已处理的特定字符串的长度。看起来状态没有保存在系统中。从这段代码可以清楚我在这里做错了什么？

【问题讨论】：

标签： java stream apache-flink flink-streaming

【解决方案1】：

进程函数使用键分区状态，这意味着每个键都有一个单独的列表。我的猜测是 10 秒内没有多个事件的键。

【讨论】：

【解决方案2】：

你的ProcessFunctionClass需要扩展Flink ProcessFunction。

【讨论】：

确实有，我忘了放这里。已编辑。