为什么Flink不重试操作员故障？

Question

我计划将我们的一个Spark应用程序迁移到Apache Flink。我试图了解它的容错特性。

我执行了以下代码，我看不到Flink实际上尝试重试任何任务（或子任务）。这可能会导致我丢失数据。我该怎么做才能确保每一次失败都能被Flink所覆盖？

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStateBackend(new FsStateBackend("file:///my-path", false))
    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
      3, // number of restart attempts
      Time.of(0, TimeUnit.SECONDS) // delay
    ))
    env.enableCheckpointing(10L)
    val text = env.socketTextStream(hostName, port)
    text
      .map { input =>
        List(input)
      }.setParallelism(1)
      .flatMap { input =>
        println(s"running for $input")
        List.fill(5)(input.head) ::: {
          println("throw exception here")
          throw new RuntimeException("some exception")
          List("c")
        }
      }

我希望在屏幕上多次看到抛出异常消息。但是，当我使用fixedDelayRestart时，它似乎只是忽略了此消息，并为其他消息继续。

Answer 1

这取决于您如何启动应用程序。

我假设您正在IDE中运行这个。在这种情况下，StreamExecutionEnvironment。getExecutionEnvironment（getExecutionEnvironment）html" target="_blank">返回一个本地流执行环境（LocalStreamExecutionEnvironment），该环境在一个进程中运行程序和所有Flink，即主进程（在Flink JobManager中）和工作进程（TaskManager）作为同一JVM进程中的线程启动。异常终止此单个进程。因此，没有剩下可以重新启动程序的Flink进程。

如果您想以容错方式运行程序，您需要将其提交到Flink环境，例如在您的本地计算机上运行的环境。下载Flink发行版，提取存档文件，然后运行./bin/start-cluster.sh。这将启动两个进程，一个主进程和一个工作进程。然后，您可以通过创建一个带有StreamExecutionE的远程执行环境来将程序提交到集群nvironment.create远程执行环境并将主机名和端口作为参数传递（请查看留档了解详细信息）。

请注意，异常仍将终止工作进程。因此，为了能够重新启动程序，您需要手动启动工作进程。在生产环境中，这通常由Kubernetes、纱线或Mesos负责。

顺便说一下，我们最近在Flink留档中添加了一个操作游乐场。这是一个基于Docker的沙盒环境，可以使用Flink的容错功能。我建议查看：Flink操作游乐场。

更多提示：

• 10ms的检查点间隔非常短。
• 文本套接字源不提供至少一次（或精确一次）保证。记录最多处理一次。