Flink 作业不能在批处理作业中使用保存点答案

【问题标题】：Flink job cant use savepoint in a batch jobFlink 作业不能在批处理作业中使用保存点
【发布时间】：2021-08-03 06:57:54
【问题描述】：

让我从一个通用的方式开始，看看我是否遗漏了一些概念：我有一个流式 flink 作业，我从中创建了一个保存点。这项工作的简化版如下所示

伪代码：

val flink = StreamExecutionEnvironment.getExecutionEnvironment
val stream = if (batchMode) {
  flink.readFile(path)
}
else {
  flink.addKafkaSource(topicName)
} 

stream.keyBy(key)
stream.process(new ProcessorWithKeyedState())

CassandraSink.addSink(stream)

只要我在没有保存点的情况下运行作业，它就可以正常工作。如果我从保存点开始工作，我会得到一个看起来像这样的异常

Caused by: java.lang.UnsupportedOperationException: Checkpoints are not supported in a single key state backend
    at org.apache.flink.streaming.api.operators.sorted.state.NonCheckpointingStorageAccess.resolveCheckpoint(NonCheckpointingStorageAccess.java:43)
    at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.restoreSavepoint(CheckpointCoordinator.java:1623)
    at org.apache.flink.runtime.scheduler.SchedulerBase.tryRestoreExecutionGraphFromSavepoint(SchedulerBase.java:362)
    at org.apache.flink.runtime.scheduler.SchedulerBase.createAndRestoreExecutionGraph(SchedulerBase.java:292)
    at org.apache.flink.runtime.scheduler.SchedulerBase.<init>(SchedulerBase.java:249)

如果我设置选项，我可以解决这个问题：

execution.batch-state-backend.enabled: false

但这最终会导致另一个错误：

Caused by: java.lang.IllegalArgumentException: The fraction of memory to allocate should not be 0. Please make sure that all types of managed memory consumers contained in the job are configured with a non-negative weight via `taskmanager.memory.managed.consumer-weights`.
at org.apache.flink.util.Preconditions.checkArgument(Preconditions.java:160)
at org.apache.flink.runtime.memory.MemoryManager.validateFraction(MemoryManager.java:673)
at org.apache.flink.runtime.memory.MemoryManager.computeMemorySize(MemoryManager.java:653)
at org.apache.flink.runtime.memory.MemoryManager.getSharedMemoryResourceForManagedMemory(MemoryManager.java:526)

当然我尝试设置配置键taskmanager.memory.managed.consumer-weights（使用DATAPROC:70,PYTHON:30）但这似乎没有任何效果。

所以我想知道我是否有概念错误并且无法在批处理作业中重用流作业中的保存点，或者我的配置是否存在问题。有什么提示吗？

【问题讨论】：

标签： scala apache-flink flink-streaming flink-batch

【解决方案1】：

在来自 flink 用户组的提示后，事实证明不可能重用来自流式作业的保存点 (https://ci.apache.org/projects/flink/flink-docs-master/docs/dev/datastream/execution_mode/#state-backends--state)。因此，我没有像批处理模式 (flink.setRuntimeMode(RuntimeExecutionMode.BATCH)) 那样运行作业，而是以默认执行模式 (STREAMING) 运行它。这有一个小缺点，它将永远运行，并且在处理完所有数据后必须由某人停止。

【讨论】：