如何在 flink 中处理 HyperLogLogPlus 的 ValueState

Question

当使用不是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);
        }
    }
}

定义值状态：

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)

Answer 1

HyperLogLogPlus是可序列化的，并且具有一个Builder对象，可以将字节数组表示形式反序列化为HyperLogLogPlus对象。因此，您可以：

1. 通过调用 .getBytes() 获取HyperLogLogPlus对象的字节数组表示形式。
2. 使用 kryo 存储字节数组
3. 在反序列化时，调用Builder.build(byte[])并接收HyperLogLogPlus对象。