重新运行后闪烁状态为空(已重新初始化)

Question

我正在尝试连接两个流，第一个是持久化MapValueState：RocksDB将数据保存在检查点文件夹中，但在新运行后，state为空。我在本地和flink集群中运行它，在集群中取消提交，然后简单地在本地重新运行

env.setStateBackend(new RocksDBStateBackend(..)
env.enableCheckpointing(1000)
...
val productDescriptionStream: KeyedStream[ProductDescription, String] = env.addSource(..)
.keyBy(_.id)
val productStockStream: KeyedStream[ProductStock, String] = env.addSource(..)
.keyBy(_.id)

和

productDescriptionStream
.connect(productStockStream)
.process(ProductProcessor())
.setParallelism(1)
env.execute("Product aggregator")

产品处理器

case class ProductProcessor() extends CoProcessFunction[ProductDescription, ProductStock, Product]{
private[this] lazy val stateDescriptor: MapStateDescriptor[String, ProductDescription] =
new MapStateDescriptor[String, ProductDescription](
"productDescription",
createTypeInformation[String],
createTypeInformation[ProductDescription]
)
private[this] lazy val states: MapState[String, ProductDescription] = getRuntimeContext.getMapState(stateDescriptor)
override def processElement1(value: ProductDescription,
ctx: CoProcessFunction[ProductDescription, ProductStock, Product]#Context,out: Collector[Product]
): Unit = {
states.put(value.id, value)
}}
override def processElement2(value: ProductStock,
ctx: CoProcessFunction[ProductDescription, ProductStock, Product]#Context, out: Collector[Product]
): Unit = {
if (states.contains(value.id)) {
val product =Product(
id = value.id,
description = Some(states.get(value.id).description),
stock = Some(value.stock),
updatedAt = value.updatedAt)
out.collect(product )
}}

Answer 1

Flink创建检查点是为了从故障中恢复，而不是在手动关闭后恢复。当作业被取消时，Flink的默认行为是删除检查点。由于作业不能再失败，因此不需要恢复。

你有几个选择：

(1( 配置检查点以在作业取消时保留检查点：

CheckpointConfig config = env.getCheckpointConfig();
config.enableExternalizedCheckpoints(
CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

然后，当你重新启动作业时，你需要表明你希望它从一个特定的检查点重新启动：

flink run -s <checkpoint-path> ...

否则，无论何时启动作业，它都将以空状态后端开始。

(2( 不要取消作业，而是使用带有保存点的stop：

flink stop [-p targetDirectory] [-d] <jobID>

之后，您将再次需要使用flink run -s ...从保存点恢复。

使用保存点停止是一种比依赖最近的检查点更干净的方法。

(3( 或者，您可以使用Ververica平台社区版，它将抽象级别提高到不必自己管理这些细节的程度。