Flink保存点不保存值状态答案

【问题标题】：Flink savepoint not saving the valuestatesFlink保存点不保存值状态
【发布时间】：2021-05-13 19:34:07
【问题描述】：

我正在编写一个 Flink 程序，在重新启动新作业时无法保存我的状态变量。

我用来自 Kafka 的连接器和一个带有 valueState 变量的 RichFlatMap 制作了一个简单的程序。该变量是一个整数，每收到一条消息就加 1。

当值大约为 15 时，我创建了一个停止保存点，但是当我从该保存点恢复它时，计数器又回到 1。

Streamingjob.java：

KeyedStream<JsonNode, Object> eventsByKey = env
                .addSource(consumer).name("Producer Topic Source")
                .keyBy(e -> {...});

eventsByKey 
        .flatMap(new Test())
        .uid("test-id")

Test.java：

public class Test extends RichFlatMapFunction<JsonNode, JsonNode> {

    private transient ValueState<Integer> persistence;

    @Override
    public void flatMap(JsonNode node, Collector<JsonNode> collector) throws Exception {
        if (persistence.value() == null) persistence.update(1);
        String device_id = node.get("data").get("device_id").toString();
        System.out.println();
        System.out.println(device_id);
        System.out.println(persistence.value());
        System.out.println();
        persistencia.update(persistence.value() + 1);
    }

    @Override
    public void open(Configuration config) {
        this.persistence = getRuntimeContext().getState(new ValueStateDescriptor<>(
                "prueba", // the state name
                Integer.class));
    }
}

这是我用于停止保存点的命令 ../bin/flink stop --savepointPath f74c92af01ed51af94e530ee0e208d7c

还有这个用于开始保存点 ../bin/flink run flink-andy-12.3.0.jar --savepointPath file:/{...}/savepoint-f74c92-6acdb05afd11

对我应该怎么做有什么想法吗？

【问题讨论】：

标签： java apache-flink flink-streaming savepoints

【解决方案1】：

要从保存点重新启动，您需要指定--fromSavepoint，而不是--savepointPath。 (docs)

换句话说：

$ ./bin/flink run \
      --fromSavepoint /{...}/savepoint-f74c92-6acdb05afd11 \
      flink-andy-12.3.0.jar

【讨论】：