禁用检查点时Flink StreamingFileSink不会摄取到S3

Question

我想在flink中使用aws s3作为数据流的接收器。我正在使用StreamingFileSink类创建一个接收器。

我的工作不需要检查点，但是当我禁用检查点时，数据不再写入S3。

案例1：启用检查点
启用检查点后，数据将成功摄取到提到的s3路径。

案例2：检查点禁用
禁用检查点时，数据不会写入s3。
我多次尝试执行作业，但每次都得到相同的结果。我在本地机器和kubernetes集群上都面临这个问题。

object FlinkTestJob {

  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // with checkpointing enabled
    env.enableCheckpointing(100)

    // Sinks
    val streamStrings: Seq[String] =
      Seq("test1", "test2", "test3", "test4", "test5", "test6", "test7", "test8", "test9", "test10")

    val testStream = env.fromCollection(streamStrings)

    val rollingPolicy = new RollingPolicy[String, String] {

      override def shouldRollOnCheckpoint(partFileState: PartFileInfo[String]): Boolean =
        partFileState.getSize > 1

      override def shouldRollOnEvent(
          partFileState: PartFileInfo[String],
          element: String): Boolean = true

      override def shouldRollOnProcessingTime(
          partFileState: PartFileInfo[String],
          currentTime: Long): Boolean = true
    }

    val sink: StreamingFileSink[String] = StreamingFileSink
      .forRowFormat(new Path("s3a://testbucket/sink"), new SimpleStringEncoder[String]("UTF-8"))
      .withRollingPolicy(rollingPolicy)
      .build()

    testStream.addSink(sink)
    env.execute("test-job")
  }
}

当我使用“WriteEasText（“s3a://testbucket/sink”）而不是StreamingFileSink”写入s3时，无论是否启用检查点，它都可以正常工作。

Flink版本：1.8.0我想了解检查点和StreamingFileSink之间的关系
谢谢

Answer 1

要使StreamingFileSink正常工作，需要启用检查点。