0

当我使用 flink sql api 处理数据时。

重启app，sum结果没有保存在checkpoint中，还是从1开始。

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StateBackend stateBackend = new FsStateBackend("file:///D:/d_backup/github/flink-best-practice/checkpoint");
env.enableCheckpointing(1000 * 60);
env.setStateBackend(stateBackend);

Table table = tableEnv.sqlQuery(
        "select sum(area_id) " +
        "from rtc_warning_gmys " +
        "where area_id = 1 " +
        "group by character_id,area_id,group_id,platform");

//   convert the Table into a retract DataStream of Row.
//   A retract stream of type X is a DataStream<Tuple2<Boolean, X>>.
//   The boolean field indicates the type of the change.
//   True is INSERT, false is DELETE.
DataStream<Tuple2<Boolean, Row>> dsRow = tableEnv.toRetractStream(table, Row.class);
dsRow.map(new MapFunction<Tuple2<Boolean,Row>, Object>() {
    @Override
    public Object map(Tuple2<Boolean, Row> booleanRowTuple2) throws Exception {
        if(booleanRowTuple2.f0) {
            System.out.println(booleanRowTuple2.f1.toString());
            return booleanRowTuple2.f1;
        }
        return null;
    }
});

env.execute("Kafka table select");

登录为：

1 2 3 ... ... 100

重新启动应用程序它仍然启动：1 2 3 ...

我认为总和值将存储在 checkpint 文件中，并且重新启动应用程序可以从 checkpint 读取最后一个结果，例如：

101 102 103 ... 120

4

1 回答 1

3

一些可能性：

  • 作业运行的时间是否足以完成检查点？仅仅因为作业产生了输出并不意味着检查点已完成。我看到您已将检查点配置为每分钟发生一次，并且检查点可能需要一些时间才能完成。

  • 工作是如何停止的？除非它们已被外部化，否则在取消作业时会删除检查点。

  • 作业是如何重新开始的？它是从检查点恢复（自动），还是从外部检查点或保存点恢复，还是从头开始重新启动？

这种实验最容易通过命令行进行。例如，您可能会

  1. 编写一个使用检查点的应用程序，并具有重启策略（例如，env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1000, 1000))）
  2. 启动本地集群
  3. “flink run -d app.jar” 开始工作
  4. 等到至少一个检查点完成
  5. “kill -9 task-manager-PID ”导致失败
  6. “taskmanager.sh start”允许作业从检查点恢复
于 2018-12-28T11:02:15.597 回答