PK哥，我想做一个能记住上次结果的wordcount统计，比如不断的输入单词”a“,会出现(a,1), (a,2), (a,3) …，但是，当我关闭再重启它又从头统计了，又出现了(a,1)，我想要接着关闭之前统计。我已经设置了checkpoint和state。代码如下：
val env = StreamExecutionEnvironment.getExecutionEnvironment

val topic = "test"
val properties = new Properties()
properties.setProperty(“bootstrap.servers”, “localhost:9092”)
properties.setProperty(“zookeeper.connect”, “localhost:2181”)
properties.setProperty(“group.id”, “test”)
val kafkaSource = new FlinkKafkaConsumer011[String](topic, new SimpleStringSchema(), properties)

val data = env.addSource(kafkaSource)

env.setStateBackend(new RocksDBStateBackend(“file:///Users/kewei.mao/app/tmp/checkpoint”))
env.enableCheckpointing(1000)
env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)
env.getCheckpointConfig.setCheckpointTimeout(60000)
env.getCheckpointConfig.setFailOnCheckpointingErrors(true)
env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)

val counts: DataStream[(String, Int)] = data
.flatMap(.toLowerCase.split(“W+”))
.filter(.nonEmpty)
.map((_, 1))
.keyBy(0)
.sum(1)

env.execute(“Streaming WordCount”)