You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Marco Villalobos <mv...@kineteque.com> on 2021/06/28 21:55:58 UTC
State Processor API and existing state
Let's say that a job has operators with UIDs: W, X, Y, and Z, and uses
RocksDB as a backend with checkpoint data URI s3://checkpoints"
Then I stop the job with a savepoint at s3://savepoint-1.
I assume that all the data within the checkpoint are stored within the
given Savepoint. Is that assumption correct?
Then, how can I fix the state in operator with UID Y, but keep all the data
in the other operators intact?
I know how to bootstrap state with the state-processor API.
I have not figured out how to correct / augment / fix the state though.
Can somebody please explain?
RE: State Processor API and existing state
Posted by "Tan, Min" <mi...@ubs.com>.
Many thanks for your prompt reply.
Yes. They are the same operators.
What I did is just modifying the content of POJO .e.g., doubling amount fields.
I am not able to send the production code, but I will do a separate mocked project to reproduce the issue. Send you the mocked code later.
Regards,
Min
From: JING ZHANG <be...@gmail.com>
Sent: 08 July 2021 04:45
To: Tan, Min <mi...@ubs.com>
Cc: Marco Villalobos <mv...@kineteque.com>; user <us...@flink.apache.org>
Subject: [External] Re: State Processor API and existing state
Hi min,
Is the POJO state in an existed operator or a new added operator?
BTW, that would be great if you would like to give the code to reproduce the exception. I need more debug to find out the reason based on the code.
Tan, Min <mi...@ubs.com>> 于2021年7月8日周四 上午2:56写道：
Hi,
I have followed the steps below in restarting a Flink job with newly modified savepoints.
I can re start a job with new savepoints as long as the Flink states are expressed in Java primitives.
When the flink states are expressed in a POJO, my job does not get restarted. I have the following exceptions.
Any ideas? Do I need to redefine any serializers?
Thank you very much for your help in advance.
Regards,
Min
-----------------------------------------------------------------------------Flink exceptions-------------------------------------------------------------------
2021-07-07 19:02:58
java.lang.Exception: Exception while creating StreamOperatorStateContext.
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:191)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:255)
at org.apache.flink.streaming.runtime.tasks.StreamTask.initializeStateAndOpen(StreamTask.java:989)
at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$0(StreamTask.java:453)
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:94)
at org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:448)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:460)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:708)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:533)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.util.FlinkException: Could not restore keyed state backend for LegacyKeyedCoProcessOperator_086b0e2b116638fe57d2d20eb2517b22_(1/1) from any of the 1 provided restore options.
at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135)
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:304)
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:131)
... 9 more
Caused by: org.apache.flink.runtime.state.BackendBuildingException: Failed when trying to restore heap backend
at org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:116)
at org.apache.flink.runtime.state.filesystem.FsStateBackend.createKeyedStateBackend(FsStateBackend.java:529)
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:288)
at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:142)
at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:121)
... 11 more
Caused by: java.lang.IllegalStateException: Missing value for the key 'org.apache.flink.state.api.output.TaggedOperatorSubtaskState'
at org.apache.flink.util.LinkedOptionalMap.unwrapOptionals(LinkedOptionalMap.java:190)
at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializerSnapshot.restoreSerializer(KryoSerializerSnapshot.java:86)
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
at java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:546)
at java.util.stream.AbstractPipeline.evaluateToArrayNode(AbstractPipeline.java:260)
at java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:505)
at org.apache.flink.api.common.typeutils.NestedSerializersSnapshotDelegate.snapshotsToRestoreSerializers(NestedSerializersSnapshotDelegate.java:225)
at org.apache.flink.api.common.typeutils.NestedSerializersSnapshotDelegate.getRestoredNestedSerializers(NestedSerializersSnapshotDelegate.java:83)
at org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot.restoreSerializer(CompositeTypeSerializerSnapshot.java:194)
at org.apache.flink.runtime.state.StateSerializerProvider.previousSchemaSerializer(StateSerializerProvider.java:189)
at org.apache.flink.runtime.state.StateSerializerProvider.currentSchemaSerializer(StateSerializerProvider.java:164)
at org.apache.flink.runtime.state.RegisteredKeyValueStateBackendMetaInfo.getStateSerializer(RegisteredKeyValueStateBackendMetaInfo.java:136)
at org.apache.flink.runtime.state.heap.StateTable.getStateSerializer(StateTable.java:315)
at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.createStateMap(CopyOnWriteStateTable.java:54)
at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.createStateMap(CopyOnWriteStateTable.java:36)
at org.apache.flink.runtime.state.heap.StateTable.<init>(StateTable.java:98)
at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.<init>(CopyOnWriteStateTable.java:49)
at org.apache.flink.runtime.state.heap.AsyncSnapshotStrategySynchronicityBehavior.newStateTable(AsyncSnapshotStrategySynchronicityBehavior.java:41)
at org.apache.flink.runtime.state.heap.HeapSnapshotStrategy.newStateTable(HeapSnapshotStrategy.java:243)
at org.apache.flink.runtime.state.heap.HeapRestoreOperation.createOrCheckStateForMetaInfo(HeapRestoreOperation.java:185)
at org.apache.flink.runtime.state.heap.HeapRestoreOperation.restore(HeapRestoreOperation.java:152)
at org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:114)
... 15 more
From: JING ZHANG <be...@gmail.com>>
Sent: 29 June 2021 07:45
To: Marco Villalobos <mv...@kineteque.com>>
Cc: user <us...@flink.apache.org>>
Subject: [External] Re: State Processor API and existing state
Hi Marco,
> I assume that all the data within the checkpoint are stored within the given Savepoint. Is that assumption correct?
Yes
> I have not figured out how to correct / augment / fix the state though. Can somebody please explain?
Please try this way.
1. Load old savepoint file, create Savepoint obj1
2. Read state of operator with UID Y in returned Savepoint obj1 by step1
3. Create `BootstrapTransformation` based on entry point class `OperatorTransformation`, bootstrap new operator state with dataset returned by step2, correct or fix old state of operator UID Y in a `StateBotstrapFunction` or `KeyedStateBootstrapFunction`
4. Load old savepoint file, create Savepoint obj2
5. Drop the old operator with UID Y by calling `removeOperator` in returned Savepoint obj2 by step4
6. Add a new Operator with UID Y by calling `withOperator` in returned Savepoint obj2 by step4 , the first parameter is uid (Y), the second parameter is returned `BootstrapTranformation` by step 3.
7. writes out returned Savepoint obj2 by step7 to a new path
In this way, in new savepoint files, states of operator withUIDs: W,X, Z are intact, only the state of operator Y is updated.
Detailed about read/write/modify savepoint could be found in document[1]
[1] https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/libs/state_processor_api/
Best regards,
JING ZHANG
Marco Villalobos <mv...@kineteque.com>> 于2021年6月29日周二 上午6:00写道：
Let's say that a job has operators with UIDs: W, X, Y, and Z, and uses RocksDB as a backend with checkpoint data URI s3://checkpoints"
Then I stop the job with a savepoint at s3://savepoint-1.
I assume that all the data within the checkpoint are stored within the given Savepoint. Is that assumption correct?
Then, how can I fix the state in operator with UID Y, but keep all the data in the other operators intact?
I know how to bootstrap state with the state-processor API.
I have not figured out how to correct / augment / fix the state though.
Can somebody please explain?
Re: State Processor API and existing state
Posted by JING ZHANG <be...@gmail.com>.
Hi min,
Is the POJO state in an existed operator or a new added operator?
BTW, that would be great if you would like to give the code to reproduce
the exception. I need more debug to find out the reason based on the code.
Tan, Min <mi...@ubs.com> 于2021年7月8日周四 上午2:56写道：
> Hi,
>
>
>
> I have followed the steps below in restarting a Flink job with newly
> modified savepoints.
>
>
>
> I can re start a job with new savepoints as long as the Flink states are
> expressed in Java primitives.
>
> When the flink states are expressed in a POJO, my job does not get
> restarted. I have the following exceptions.
>
>
>
> Any ideas? Do I need to redefine any serializers?
>
>
>
> Thank you very much for your help in advance.
>
> Regards,
>
>
>
> Min
>
>
>
>
>
> -----------------------------------------------------------------------------Flink
> exceptions-------------------------------------------------------------------
>
> 2021-07-07 19:02:58
>
> java.lang.Exception: Exception while creating StreamOperatorStateContext.
>
> at org.apache.flink.streaming.api.operators.
> StreamTaskStateInitializerImpl.streamOperatorStateContext(
> StreamTaskStateInitializerImpl.java:191)
>
> at org.apache.flink.streaming.api.operators.AbstractStreamOperator
> .initializeState(AbstractStreamOperator.java:255)
>
> at org.apache.flink.streaming.runtime.tasks.StreamTask
> .initializeStateAndOpen(StreamTask.java:989)
>
> at org.apache.flink.streaming.runtime.tasks.StreamTask
> .lambda$beforeInvoke$0(StreamTask.java:453)
>
> at org.apache.flink.streaming.runtime.tasks.
> StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(
> StreamTaskActionExecutor.java:94)
>
> at org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(
> StreamTask.java:448)
>
> at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(
> StreamTask.java:460)
>
> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:708)
>
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:533)
>
> at java.lang.Thread.run(Thread.java:748)
>
> Caused by: org.apache.flink.util.FlinkException: Could not restore keyed
> state backend for
> LegacyKeyedCoProcessOperator_086b0e2b116638fe57d2d20eb2517b22_(1/1) from
> any of the 1 provided restore options.
>
> at org.apache.flink.streaming.api.operators.BackendRestorerProcedure
> .createAndRestore(BackendRestorerProcedure.java:135)
>
> at org.apache.flink.streaming.api.operators.
> StreamTaskStateInitializerImpl.keyedStatedBackend(
> StreamTaskStateInitializerImpl.java:304)
>
> at org.apache.flink.streaming.api.operators.
> StreamTaskStateInitializerImpl.streamOperatorStateContext(
> StreamTaskStateInitializerImpl.java:131)
>
> ... 9 more
>
> Caused by: org.apache.flink.runtime.state.BackendBuildingException: Failed
> when trying to restore heap backend
>
> at org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder
> .build(HeapKeyedStateBackendBuilder.java:116)
>
> at org.apache.flink.runtime.state.filesystem.FsStateBackend
> .createKeyedStateBackend(FsStateBackend.java:529)
>
> at org.apache.flink.streaming.api.operators.
> StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(
> StreamTaskStateInitializerImpl.java:288)
>
> at org.apache.flink.streaming.api.operators.BackendRestorerProcedure
> .attemptCreateAndRestore(BackendRestorerProcedure.java:142)
>
> at org.apache.flink.streaming.api.operators.BackendRestorerProcedure
> .createAndRestore(BackendRestorerProcedure.java:121)
>
> ... 11 more
>
> Caused by: java.lang.IllegalStateException: Missing value for the key
> 'org.apache.flink.state.api.output.TaggedOperatorSubtaskState'
>
> at org.apache.flink.util.LinkedOptionalMap.unwrapOptionals(
> LinkedOptionalMap.java:190)
>
> at org.apache.flink.api.java.typeutils.runtime.kryo.
> KryoSerializerSnapshot.restoreSerializer(KryoSerializerSnapshot.java:86)
>
> at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline
> .java:193)
>
> at java.util.Spliterators$ArraySpliterator.forEachRemaining(
> Spliterators.java:948)
>
> at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:
> 482)
>
> at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline
> .java:472)
>
> at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:
> 546)
>
> at java.util.stream.AbstractPipeline.evaluateToArrayNode(
> AbstractPipeline.java:260)
>
> at java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:
> 505)
>
> at org.apache.flink.api.common.typeutils.
> NestedSerializersSnapshotDelegate.snapshotsToRestoreSerializers(
> NestedSerializersSnapshotDelegate.java:225)
>
> at org.apache.flink.api.common.typeutils.
> NestedSerializersSnapshotDelegate.getRestoredNestedSerializers(
> NestedSerializersSnapshotDelegate.java:83)
>
> at org.apache.flink.api.common.typeutils.
> CompositeTypeSerializerSnapshot.restoreSerializer(
> CompositeTypeSerializerSnapshot.java:194)
>
> at org.apache.flink.runtime.state.StateSerializerProvider
> .previousSchemaSerializer(StateSerializerProvider.java:189)
>
> at org.apache.flink.runtime.state.StateSerializerProvider
> .currentSchemaSerializer(StateSerializerProvider.java:164)
>
> at org.apache.flink.runtime.state.
> RegisteredKeyValueStateBackendMetaInfo.getStateSerializer(
> RegisteredKeyValueStateBackendMetaInfo.java:136)
>
> at org.apache.flink.runtime.state.heap.StateTable.getStateSerializer(
> StateTable.java:315)
>
> at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable
> .createStateMap(CopyOnWriteStateTable.java:54)
>
> at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable
> .createStateMap(CopyOnWriteStateTable.java:36)
>
> at org.apache.flink.runtime.state.heap.StateTable.<init>(StateTable
> .java:98)
>
> at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.<init>(
> CopyOnWriteStateTable.java:49)
>
> at org.apache.flink.runtime.state.heap.
> AsyncSnapshotStrategySynchronicityBehavior.newStateTable(
> AsyncSnapshotStrategySynchronicityBehavior.java:41)
>
> at org.apache.flink.runtime.state.heap.HeapSnapshotStrategy
> .newStateTable(HeapSnapshotStrategy.java:243)
>
> at org.apache.flink.runtime.state.heap.HeapRestoreOperation
> .createOrCheckStateForMetaInfo(HeapRestoreOperation.java:185)
>
> at org.apache.flink.runtime.state.heap.HeapRestoreOperation.restore(
> HeapRestoreOperation.java:152)
>
> at org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder
> .build(HeapKeyedStateBackendBuilder.java:114)
>
> ... 15 more
>
>
>
>
>
>
>
> *From:* JING ZHANG <be...@gmail.com>
> *Sent:* 29 June 2021 07:45
> *To:* Marco Villalobos <mv...@kineteque.com>
> *Cc:* user <us...@flink.apache.org>
> *Subject:* [External] Re: State Processor API and existing state
>
>
>
> Hi Marco,
>
> > I assume that all the data within the checkpoint are stored within the
> given Savepoint. Is that assumption correct?
>
> Yes
>
> > I have not figured out how to correct / augment / fix the state though.
> Can somebody please explain?
>
> Please try this way.
>
> 1. Load old savepoint file, create Savepoint obj1
>
> 2. Read state of operator with UID Y in returned Savepoint obj1 by step1
>
> 3. Create `BootstrapTransformation` based on entry point class
> `OperatorTransformation`, bootstrap new operator state with dataset
> returned by step2, correct or fix old state of operator UID Y in a
> `StateBotstrapFunction` or `KeyedStateBootstrapFunction`
>
> 4. Load old savepoint file, create Savepoint obj2
>
> 5. Drop the old operator with UID Y by calling `removeOperator` in
> returned Savepoint obj2 by step4
>
> 6. Add a new Operator with UID Y by calling `withOperator` in returned
> Savepoint obj2 by step4 , the first parameter is uid (Y), the second
> parameter is returned `BootstrapTranformation` by step 3.
>
> 7. writes out returned Savepoint obj2 by step7 to a new path
>
>
>
> In this way, in new savepoint files, states of operator withUIDs: W,X, Z
> are intact, only the state of operator Y is updated.
>
> Detailed about read/write/modify savepoint could be found in document[1]
>
>
>
> [1]
> https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/libs/state_processor_api/
>
>
>
> Best regards,
>
> JING ZHANG
>
>
>
> Marco Villalobos <mv...@kineteque.com> 于2021年6月29日周二 上午6:00写道：
>
> Let's say that a job has operators with UIDs: W, X, Y, and Z, and uses
> RocksDB as a backend with checkpoint data URI s3://checkpoints"
>
>
>
> Then I stop the job with a savepoint at s3://savepoint-1.
>
>
>
> I assume that all the data within the checkpoint are stored within the
> given Savepoint. Is that assumption correct?
>
>
>
> Then, how can I fix the state in operator with UID Y, but keep all the
> data in the other operators intact?
>
>
>
> I know how to bootstrap state with the state-processor API.
>
> I have not figured out how to correct / augment / fix the state though.
>
>
>
> Can somebody please explain?
>
>
RE: State Processor API and existing state
Posted by "Tan, Min" <mi...@ubs.com>.
Hi,
I have followed the steps below in restarting a Flink job with newly modified savepoints.
I can re start a job with new savepoints as long as the Flink states are expressed in Java primitives.
When the flink states are expressed in a POJO, my job does not get restarted. I have the following exceptions.
Any ideas? Do I need to redefine any serializers?
Thank you very much for your help in advance.
Regards,
Min
-----------------------------------------------------------------------------Flink exceptions-------------------------------------------------------------------
2021-07-07 19:02:58
java.lang.Exception: Exception while creating StreamOperatorStateContext.
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:191)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:255)
at org.apache.flink.streaming.runtime.tasks.StreamTask.initializeStateAndOpen(StreamTask.java:989)
at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$0(StreamTask.java:453)
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:94)
at org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:448)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:460)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:708)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:533)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.util.FlinkException: Could not restore keyed state backend for LegacyKeyedCoProcessOperator_086b0e2b116638fe57d2d20eb2517b22_(1/1) from any of the 1 provided restore options.
at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135)
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:304)
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:131)
... 9 more
Caused by: org.apache.flink.runtime.state.BackendBuildingException: Failed when trying to restore heap backend
at org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:116)
at org.apache.flink.runtime.state.filesystem.FsStateBackend.createKeyedStateBackend(FsStateBackend.java:529)
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:288)
at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:142)
at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:121)
... 11 more
Caused by: java.lang.IllegalStateException: Missing value for the key 'org.apache.flink.state.api.output.TaggedOperatorSubtaskState'
at org.apache.flink.util.LinkedOptionalMap.unwrapOptionals(LinkedOptionalMap.java:190)
at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializerSnapshot.restoreSerializer(KryoSerializerSnapshot.java:86)
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
at java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:546)
at java.util.stream.AbstractPipeline.evaluateToArrayNode(AbstractPipeline.java:260)
at java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:505)
at org.apache.flink.api.common.typeutils.NestedSerializersSnapshotDelegate.snapshotsToRestoreSerializers(NestedSerializersSnapshotDelegate.java:225)
at org.apache.flink.api.common.typeutils.NestedSerializersSnapshotDelegate.getRestoredNestedSerializers(NestedSerializersSnapshotDelegate.java:83)
at org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot.restoreSerializer(CompositeTypeSerializerSnapshot.java:194)
at org.apache.flink.runtime.state.StateSerializerProvider.previousSchemaSerializer(StateSerializerProvider.java:189)
at org.apache.flink.runtime.state.StateSerializerProvider.currentSchemaSerializer(StateSerializerProvider.java:164)
at org.apache.flink.runtime.state.RegisteredKeyValueStateBackendMetaInfo.getStateSerializer(RegisteredKeyValueStateBackendMetaInfo.java:136)
at org.apache.flink.runtime.state.heap.StateTable.getStateSerializer(StateTable.java:315)
at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.createStateMap(CopyOnWriteStateTable.java:54)
at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.createStateMap(CopyOnWriteStateTable.java:36)
at org.apache.flink.runtime.state.heap.StateTable.<init>(StateTable.java:98)
at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.<init>(CopyOnWriteStateTable.java:49)
at org.apache.flink.runtime.state.heap.AsyncSnapshotStrategySynchronicityBehavior.newStateTable(AsyncSnapshotStrategySynchronicityBehavior.java:41)
at org.apache.flink.runtime.state.heap.HeapSnapshotStrategy.newStateTable(HeapSnapshotStrategy.java:243)
at org.apache.flink.runtime.state.heap.HeapRestoreOperation.createOrCheckStateForMetaInfo(HeapRestoreOperation.java:185)
at org.apache.flink.runtime.state.heap.HeapRestoreOperation.restore(HeapRestoreOperation.java:152)
at org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:114)
... 15 more
From: JING ZHANG <be...@gmail.com>
Sent: 29 June 2021 07:45
To: Marco Villalobos <mv...@kineteque.com>
Cc: user <us...@flink.apache.org>
Subject: [External] Re: State Processor API and existing state
Hi Marco,
> I assume that all the data within the checkpoint are stored within the given Savepoint. Is that assumption correct?
Yes
> I have not figured out how to correct / augment / fix the state though. Can somebody please explain?
Please try this way.
1. Load old savepoint file, create Savepoint obj1
2. Read state of operator with UID Y in returned Savepoint obj1 by step1
3. Create `BootstrapTransformation` based on entry point class `OperatorTransformation`, bootstrap new operator state with dataset returned by step2, correct or fix old state of operator UID Y in a `StateBotstrapFunction` or `KeyedStateBootstrapFunction`
4. Load old savepoint file, create Savepoint obj2
5. Drop the old operator with UID Y by calling `removeOperator` in returned Savepoint obj2 by step4
6. Add a new Operator with UID Y by calling `withOperator` in returned Savepoint obj2 by step4 , the first parameter is uid (Y), the second parameter is returned `BootstrapTranformation` by step 3.
7. writes out returned Savepoint obj2 by step7 to a new path
In this way, in new savepoint files, states of operator withUIDs: W,X, Z are intact, only the state of operator Y is updated.
Detailed about read/write/modify savepoint could be found in document[1]
[1] https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/libs/state_processor_api/
Best regards,
JING ZHANG
Marco Villalobos <mv...@kineteque.com>> 于2021年6月29日周二 上午6:00写道：
Let's say that a job has operators with UIDs: W, X, Y, and Z, and uses RocksDB as a backend with checkpoint data URI s3://checkpoints"
Then I stop the job with a savepoint at s3://savepoint-1.
I assume that all the data within the checkpoint are stored within the given Savepoint. Is that assumption correct?
Then, how can I fix the state in operator with UID Y, but keep all the data in the other operators intact?
I know how to bootstrap state with the state-processor API.
I have not figured out how to correct / augment / fix the state though.
Can somebody please explain?
Re: State Processor API and existing state
Posted by JING ZHANG <be...@gmail.com>.
Hi Marco,
> I assume that all the data within the checkpoint are stored within the
given Savepoint. Is that assumption correct?
Yes
> I have not figured out how to correct / augment / fix the state though.
Can somebody please explain?
Please try this way.
1. Load old savepoint file, create Savepoint obj1
2. Read state of operator with UID Y in returned Savepoint obj1 by step1
3. Create `BootstrapTransformation` based on entry point class
`OperatorTransformation`, bootstrap new operator state with dataset
returned by step2, correct or fix old state of operator UID Y in a
`StateBotstrapFunction` or `KeyedStateBootstrapFunction`
4. Load old savepoint file, create Savepoint obj2
5. Drop the old operator with UID Y by calling `removeOperator` in returned
Savepoint obj2 by step4
6. Add a new Operator with UID Y by calling `withOperator` in returned
Savepoint obj2 by step4 , the first parameter is uid (Y), the second
parameter is returned `BootstrapTranformation` by step 3.
7. writes out returned Savepoint obj2 by step7 to a new path
In this way, in new savepoint files, states of operator withUIDs: W,X, Z
are intact, only the state of operator Y is updated.
Detailed about read/write/modify savepoint could be found in document[1]
[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/libs/state_processor_api/
Best regards,
JING ZHANG
Marco Villalobos <mv...@kineteque.com> 于2021年6月29日周二 上午6:00写道：
> Let's say that a job has operators with UIDs: W, X, Y, and Z, and uses
> RocksDB as a backend with checkpoint data URI s3://checkpoints"
>
> Then I stop the job with a savepoint at s3://savepoint-1.
>
> I assume that all the data within the checkpoint are stored within the
> given Savepoint. Is that assumption correct?
>
> Then, how can I fix the state in operator with UID Y, but keep all the
> data in the other operators intact?
>
> I know how to bootstrap state with the state-processor API.
> I have not figured out how to correct / augment / fix the state though.
>
> Can somebody please explain?
>