我正在从一个无界数据源(Kafka)读取数据,并将其词频写入到另一个Kafka主题中。现在,我想在Beam Pipeline中执行checkpoint。我已经按照Apache Beam文档中的所有说明进行了操作,但是checkpoint目录仍未创建。
以下是我用于Pipeline的参数:
--runner=FlinkRunner
--streaming=true
--parallelism=2
--checkpointingInterval=1000
--checkpointTimeoutMillis=5000
--minPauseBetweenCheckpoints=500
--externalizedCheckpointsEnabled=true
--retainExternalizedCheckpointsOnCancellation=true
有人可以帮我解决一下检查点吗?