【发布时间】：2021-07-02 14:50:49
【问题描述】：
我有一个用 Flink (Scala) 编写的示例流 WordCount 示例。在其中，我想使用外部检查点来在发生故障时进行恢复。但它没有按预期工作。
我的代码如下：
object WordCount {
def main(args: Array[String]) {
// set up the execution environment
val env = StreamExecutionEnvironment
.getExecutionEnvironment
.setStateBackend(new RocksDBStateBackend("file:///path/to/checkpoint", true))
// start a checkpoint every 1000 ms
env.enableCheckpointing(1000)
// set mode to exactly-once (this is the default)
env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
// make sure 500 ms of progress happen between checkpoints
env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)
// checkpoints have to complete within one minute, or are discarded
env.getCheckpointConfig.setCheckpointTimeout(60000)
// prevent the tasks from failing if an error happens in their checkpointing, the checkpoint will just be declined.
env.getCheckpointConfig.setFailOnCheckpointingErrors(false)
// allow only one checkpoint to be in progress at the same time
env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
// prepare Kafka consumer properties
val kafkaConsumerProperties = new Properties
kafkaConsumerProperties.setProperty("zookeeper.connect", "localhost:2181")
kafkaConsumerProperties.setProperty("group.id", "flink")
kafkaConsumerProperties.setProperty("bootstrap.servers", "localhost:9092")
// set up Kafka Consumer
val kafkaConsumer = new FlinkKafkaConsumer[String]("input", new SimpleStringSchema, kafkaConsumerProperties)
println("Executing WordCount example.")
// get text from Kafka
val text = env.addSource(kafkaConsumer)
val counts: DataStream[(String, Int)] = text
// split up the lines in pairs (2-tuples) containing: (word,1)
.flatMap(_.toLowerCase.split("\\W+"))
.filter(_.nonEmpty)
.map((_, 1))
// group by the tuple field "0" and sum up tuple field "1"
.keyBy(0)
.mapWithState((in: (String, Int), count: Option[Int]) =>
count match {
case Some(c) => ((in._1, c), Some(c + in._2))
case None => ((in._1, 1), Some(in._2 + 1))
})
// emit result
println("Printing result to stdout.")
counts.print()
// execute program
env.execute("Streaming WordCount")
}
}
我第一次运行程序后得到的输出是：
(hi, 1)
(hi, 2)
我第二次运行程序后得到的输出是：
(hi, 1)
我的期望是第二次运行程序应该给我以下输出：
(hi, 3)
由于我是 Apache Flink 的新手，我不知道如何达到预期的效果。谁能帮我实现正确的行为？
【问题讨论】：
-
你如何停止应用程序？
-
@DominikWosiński - 我正在通过 IntelliJ Idea 运行主课。因此，运行和停止程序的命令分别是：Ctrl + Shift + F10 和 Ctrl + F2。
标签： apache-flink flink-streaming checkpointing