将 uid 添加到运算符会导致 "Cannot map checkpoint/savepoint state for operator"

Question

我有一个简单的管道

env.addSource(kafkaConsumer).uid("kafka-src").name(consumerName)
    .keyBy(_.id)
    .process(new Processor).uid("processor")
    .addSink(kafkaProducer).name(producerName)

现在我尝试像这样简单地将 uid 添加到接收器

env.addSource(kafkaConsumer).uid("kafka-src").name(consumerName)
    .keyBy(_.id)
    .process(new Processor).uid("processor")
    .addSink(kafkaProducer).name(producerName).uid("kafka-sink")

但我收到很长的异常，这似乎是消息的一部分：

Caused by: java.lang.IllegalStateException: Failed to rollback to checkpoint/savepoint file:/tmp/rocksdb/savepoint-445173-011657873d74. Cannot map checkpoint/savepoint state for operator 3cfeb06db0484d5556a7de8db2025f09 to the new program, because the operator is not available in the new program. If you want to allow to skip this, you can set the --allowNonRestoredState option on the CLI.
    at org.apache.flink.runtime.checkpoint.Checkpoints.loadAndValidateCheckpoint(Checkpoints.java:205)
    at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.restoreSavepoint(CheckpointCoordinator.java:1103)
    at org.apache.flink.runtime.jobmaster.JobMaster.tryRestoreExecutionGraphFromSavepoint(JobMaster.java:1251)
    at org.apache.flink.runtime.jobmaster.JobMaster.createAndRestoreExecutionGraph(JobMaster.java:1175)
    at org.apache.flink.runtime.jobmaster.JobMaster.<init>(JobMaster.java:299)
    at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.createJobMasterService(DefaultJobMasterServiceFactory.java:83)
    at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.createJobMasterService(DefaultJobMasterServiceFactory.java:37)
    at org.apache.flink.runtime.jobmaster.JobManagerRunner.<init>(JobManagerRunner.java:146)

这有意义吗？有没有办法在不丢失保存点的情况下解决它？

Answer 1

这个问题是有道理的，因为如果您不手动指定 ID，它们将自动生成。大概生成的ID是3cfeb06db0484d5556a7de8db2025f09.

你有三个选择：

在没有保存点的情况下启动作业：这将导致数据丢失；
运行允许non-restored状态的工作；
使用 3cfeb06db0484d5556a7de8db2025f09 作为您的运营商的 uid。

以下链接可能对您有所帮助：

将 uid 添加到运算符会导致 "Cannot map checkpoint/savepoint state for operator"

adding uid to operator causes "Cannot map checkpoint/savepoint state for operator"

apache-flink

flink-streaming