Flink状态模式迁移

Question

我在一个使用MemoryStateBackend的独立集群上有一个flink流应用程序。Kryo的TaggedFieldSerializer被用作默认的序列化程序。

当我更改状态的模式并重新部署应用程序时，我得到了以下异常

Caused by: org.apache.flink.util.StateMigrationException: State migration isn't supported, yet.
at org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.tryRegisterStateTable(HeapKeyedStateBackend.java:209)
at org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.tryRegisterStateTable(HeapKeyedStateBackend.java:142)
at org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.createValueState(HeapKeyedStateBackend.java:234)
at org.apache.flink.runtime.state.AbstractKeyedStateBackend$1.createValueState(AbstractKeyedStateBackend.java:315)
at org.apache.flink.api.common.state.ValueStateDescriptor.bind(ValueStateDescriptor.java:128)
at org.apache.flink.api.common.state.ValueStateDescriptor.bind(ValueStateDescriptor.java:35)
at org.apache.flink.runtime.state.AbstractKeyedStateBackend.getOrCreateKeyedState(AbstractKeyedStateBackend.java:312)
at org.apache.flink.runtime.state.AbstractKeyedStateBackend.getPartitionedState(AbstractKeyedStateBackend.java:392)
at org.apache.flink.runtime.state.DefaultKeyedStateStore.getPartitionedState(DefaultKeyedStateStore.java:124)
at org.apache.flink.runtime.state.DefaultKeyedStateStore.getState(DefaultKeyedStateStore.java:60)

如果有人建议我绕过这个问题，或者我应该使用FsStateBackend来解决这个问题，这将是非常有帮助的。

如果我想对运行在独立集群上的flink应用程序使用FsStateBackend on S3，那么必须做哪些配置更改。

Answer 1

使用FsStateBackend并不能解决这个问题，因为它还会在幕后使用HeapKeyedStateBackend，这就是抛出这个异常的原因。

有翻转-22来帮助解决这个状态迁移问题，但它还没有实现。

目前我听说过的最好的选择是使用基于Avro的序列化程序，因为它可以被实现，从而无缝地处理旧的和新的模式。但这不是给那些胆小的人用的。

关于FsStateBackend配置，请参阅(编写良好的)文档这里。