apache-flink - flink中如何处理HyperLogLogPlus的ValueState
问题描述
当使用不是POJO的java对象的ValueState时,如HyperLogLogPlus,如何处理序列化问题?
我尝试定义 HyperLogLogPlusSerializer 扩展 TypeSerializer,代码为:
public class HyperLogLogPlusSerializer extends TypeSerializer<HyperLogLogPlus> {
private static Logger logger = LoggerFactory.getLogger(HyperLogLogPlusSerializer.class);
private static final long serialVersionUID = 1L;
@Override
public boolean isImmutableType() {
return false;
}
@Override
public TypeSerializer<HyperLogLogPlus> duplicate() {
logger.info("duplicate invoked.");
return this;
}
@Override
public HyperLogLogPlus createInstance() {
return new HyperLogLogPlus(20,25);
}
@Override
public HyperLogLogPlus copy(HyperLogLogPlus from) {
try{
byte[] bytes = from.getBytes();
return HyperLogLogPlus.Builder.build(bytes);
}catch (Exception e){
}
return null;
}
@Override
public HyperLogLogPlus copy(HyperLogLogPlus from, HyperLogLogPlus reuse) {
return copy(from);
}
@Override
public int getLength() {
return -1;
}
@Override
public void serialize(HyperLogLogPlus record, DataOutputView target) throws IOException {
byte[] bytes = record.getBytes();
target.writeInt(bytes.length);
target.write(bytes);
logger.info("serialize:{}", bytes.length);
}
@Override
public HyperLogLogPlus deserialize(DataInputView source) throws IOException {
int dataLength = source.readInt();
byte[] bytes = new byte[dataLength];
source.read(bytes);
logger.info("deserialize:{}", bytes.length);
return HyperLogLogPlus.Builder.build(bytes);
}
@Override
public HyperLogLogPlus deserialize(HyperLogLogPlus reuse, DataInputView source) throws IOException {
return deserialize(source);
}
@Override
public void copy(DataInputView source, DataOutputView target) throws IOException {
int dataLength = source.readInt();
byte[] bytes = new byte[dataLength];
source.read(bytes);
target.writeInt(dataLength);
target.write(bytes);
}
@Override
public boolean equals(Object obj) {
if(obj instanceof HyperLogLogPlusSerializer){
return true;
}else{
return false;
}
}
@Override
public boolean canEqual(Object obj) {
return obj instanceof HyperLogLogPlusSerializer;
}
@Override
public int hashCode() {
return this.getClass().hashCode();
}
@Override
public TypeSerializerSnapshot<HyperLogLogPlus> snapshotConfiguration() {
logger.error("snapshotConfiguration invoked.");
return new HyperLogLogPlusSerializerSnapshot();
}
public static final class HyperLogLogPlusSerializerSnapshot extends SimpleTypeSerializerSnapshot<HyperLogLogPlus> {
public HyperLogLogPlusSerializerSnapshot(){
super(HyperLogLogPlusSerializer.class);
}
}
}
ValueState 定义:
private ValueState<HyperLogLogPlus> showUVState;
showUVState = getRuntimeContext().getState(new ValueStateDescriptor("showUVState",new HyperLogLogPlusSerializer()));
flink 作业使用检查点正常运行,但从保存点恢复时抛出异常:
Caused by: org.apache.flink.util.FlinkException: Could not restore keyed state backend for LegacyKeyedProcessOperator_306d8342cb5b2ad8b53f1be57f65bee8_(28/32) from any of the 1 provided restore options.
at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:137)
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:284)
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:135)
... 5 more
Caused by: java.lang.NullPointerException
at org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.readKeyGroupStateData(HeapKeyedStateBackend.java:479)
at org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.readStateHandleStateData(HeapKeyedStateBackend.java:453)
at org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.restorePartitionedState(HeapKeyedStateBackend.java:410)
at org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.restore(HeapKeyedStateBackend.java:358)
at org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.restore(HeapKeyedStateBackend.java:104)
at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:151)
at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:123)
解决方案
HyperLogLogPlus
是可序列化的,并且有一个Builder
可以将字节数组表示反序列化为对象的HyperLogLogPlus
对象。因此,您可以:
HyperLogLogPlus
通过调用获取对象的字节数组表示.getBytes()
。- 用 kryo 存储字节数组
- 在反序列化时,调用
Builder.build(byte[])
并接收一个HyperLogLogPlus
对象。
推荐阅读
- azure - 如何为 AzureDevOps 设置 AzureDevOps 代理?
- java - 有没有办法我可以在 nlogn 复杂性中解决这个问题
- amadeus - 新的 v2 航班优惠端点包括航空公司代码和排除航空公司代码不能一起使用吗?
- r - 基于多个值排序堆叠条形图
- mongodb - Positional Projection 中的多个位置运算符
- python - 在某个点将列表分解为更小的列表
- python - 如何使用 python 在 Microsoft Graph 中显示用户的图像
- html - v-text-field 中的中心标签
- typescript - 如何告诉 TypeScript 我的函数参数是 Enum 键?
- javascript - 使用导致所有文档返回的 IN 子句时,Firebase 查询过滤器未定义