【问题标题】：Serialize Guava's MinMaxPriorityQueue序列化 Guava 的 MinMaxPriorityQueue
【发布时间】：2018-06-28 15:37:39
【问题描述】：

在研究了我的Flink application is not working properly 的原因几天后，我得出结论，问题在于我正在使用的MinMaxPriorityQueue。

看来这个结构是不可序列化的。我尝试了几种方法来序列化它：

env.getConfig.registerTypeWithKryoSerializer(classOf[MinMaxPriorityQueue[Double]], classOf[JavaSerializer])
env.getConfig.registerTypeWithKryoSerializer(classOf[MinMaxPriorityQueue[java.lang.Double]], classOf[ProtobufSerializer]);
env.getConfig().addDefaultKryoSerializer(MyCustomType.class, TBaseSerializer.class);

他们都没有运气。

但是我发现了这个：Serializing Guava's ImmutableTable

是否有等效于 MinMaxPriorityQueue 的方法，或者序列化的方法？

更新

我已将 Tomasz 翻译成 scala：

class MinMaxPriorityQueueSerializer extends Serializer[MinMaxPriorityQueue[Object]] {
  private[this] val log = LoggerFactory.getLogger(this.getClass)
  setImmutable(false)
  setAcceptsNull(false)

  val OPTIMIZE_POSITIVE = true

  override def read(kryo: Kryo, input: Input, aClass: Class[MinMaxPriorityQueue[Object]]): MinMaxPriorityQueue[Object] = {
    log.error("Kryo READ")
    val comparator: Ordering[Object] = kryo.readClassAndObject(input).asInstanceOf[Ordering[Object]]
    val size = input.readInt(OPTIMIZE_POSITIVE)

    val queue: MinMaxPriorityQueue[Object] = MinMaxPriorityQueue.orderedBy(comparator)
      .expectedSize(size)
      .create()

    (0 to size).foreach(_ => queue.offer(kryo.readClassAndObject(input)))

    queue
  }

  override def write(kryo: Kryo, output: Output, queue: MinMaxPriorityQueue[Object]): Unit = {
    log.error("Kryo WRITE")
    kryo.writeClassAndObject(output, queue.comparator)

    val declaredSize = queue.size
    output.writeInt(declaredSize, OPTIMIZE_POSITIVE)

    val actualSize = queue.toArray.foldLeft(0) {
      case (z, q) =>
        kryo.writeClassAndObject(output, q)
        z + 1
    }

    Preconditions.checkState(
      declaredSize == actualSize,
      "Declared size (%s) different than actual size (%s)", declaredSize, actualSize)
  }
}

并在 flink 中设置 kryo 以使用该序列化器：

env.getConfig.addDefaultKryoSerializer(classOf[MinMaxPriorityQueue[Double]], classOf[MinMaxPriorityQueueSerializer])       

env.getConfig.registerTypeWithKryoSerializer(classOf[MinMaxPriorityQueue[Double]], classOf[MinMaxPriorityQueueSerializer])

但它似乎永远不会被调用，因为我在日志中的任何地方都看不到 log.error("Kryo READ") 和 log.error("Kryo WRITE") 的输出

即使我正在更新它，转换仍然返回一个空的 MinMaxPriorityQueue。

更新 2

我已经实现了 SerializerTester，但我得到了一个 bufferUnderflow：

object Main {

  def main(args: Array[String]) {

    val tester = new MinMaxPriorityQueueSerializerTester()

    val inQueue: MinMaxPriorityQueue[java.lang.Double] = MinMaxPriorityQueue.create()
    inQueue.add(1.0)

    val outputStream = new ByteArrayOutputStream()
    tester.serialize(outputStream, inQueue)

    val inputStream = new ByteArrayInputStream(outputStream.toByteArray())
    val outQueue: MinMaxPriorityQueue[java.lang.Double] = tester.deserialize(inputStream);

    System.out.println(inQueue);
    System.out.println(outQueue);

  }

  class MinMaxPriorityQueueSerializerTester {
    val kryo = new Kryo
    kryo.setInstantiatorStrategy(new StdInstantiatorStrategy)
    registerMinMaxSerializer();
    //  allowForClassesWithoutNoArgConstructor(); // needed to serialize Ordering

    def registerMinMaxSerializer() {
      kryo.addDefaultSerializer(classOf[MinMaxPriorityQueue[java.lang.Double]], new MinMaxPriorityQueueSerializer());
    }

    def serialize(out: OutputStream, queue: MinMaxPriorityQueue[java.lang.Double]) {
      // try (Output output = new Output(out)) {
      val output = new Output(out)
      kryo.writeClassAndObject(output, queue)
      //      kryo.writeObject(output, queue)
      //}
      output.flush
    }

    def deserialize(in: InputStream): MinMaxPriorityQueue[java.lang.Double] = {
      //try (Input input = new Input(in)) {
      val input = new Input(in)
      //kryo.readObject(input, classOf[MinMaxPriorityQueue[java.lang.Double]])
      kryo.readClassAndObject(input).asInstanceOf[MinMaxPriorityQueue[java.lang.Double]]
      //p}
    }
  }

【问题讨论】：

您可以在 Guava 的 Github 存储库中创建问题并请求该功能。有issue #615，里面提到了MinMaxPriorityQueue，但最后没有包含在ticket的范围内。
@Xaerxess，谢谢，我已经打开了：github.com/google/guava/issues/3192
@elbaulp 你能发布 bufferUnderflow 的堆栈跟踪吗？
没关系，我改用了 intervalHeap 的这个实现：github.com/allenbh/gkutil_java/blob/master/src/gkimfl/util/…，因为它开箱即用。谢谢你的帮助，我接受你的回答。
@elbaulp 谢谢，尽管我的回答并没有真正解决您的问题。您可以考虑发布一个单独的答案，说明您是如何解决的，以防其他人遇到类似问题。

标签： scala guava apache-flink kryo

【解决方案1】：

您可以使用自定义 Kryo Serializer。

这是一个示例（用 Java 编写）：

class MinMaxPriorityQueueSerializer extends Serializer<MinMaxPriorityQueue<Object>> {

    private static final boolean OPTIMIZE_POSITIVE = true;

    protected MinMaxPriorityQueueSerializer() {
        setAcceptsNull(false);
        setImmutable(false);
    }

    @Override
    public void write(Kryo kryo, Output output, MinMaxPriorityQueue<Object> queue) {
        kryo.writeClassAndObject(output, queue.comparator());

        int declaredSize = queue.size();
        output.writeInt(declaredSize, OPTIMIZE_POSITIVE);

        int actualSize = 0;
        for (Object element : queue) {
            kryo.writeClassAndObject(output, element);
            actualSize++;
        }

        Preconditions.checkState(
                declaredSize == actualSize,
                "Declared size (%s) different than actual size (%s)", declaredSize, actualSize
        );
    }

    @Override
    public MinMaxPriorityQueue<Object> read(Kryo kryo, Input input, Class<MinMaxPriorityQueue<Object>> type) {
        @SuppressWarnings("unchecked")
        Comparator<Object> comparator = (Comparator<Object>) kryo.readClassAndObject(input);
        int size = input.readInt(OPTIMIZE_POSITIVE);

        MinMaxPriorityQueue<Object> queue = MinMaxPriorityQueue.orderedBy(comparator)
                .expectedSize(size)
                .create();

        for (int i = 0; i < size; ++i) {
            queue.offer(kryo.readClassAndObject(input));
        }
        return queue;
    }
}

你可以这样使用它：

class MinMaxPriorityQueueSerializerTester {

    public static void main(String[] args) {
        MinMaxPriorityQueueSerializerTester tester = new MinMaxPriorityQueueSerializerTester();

        MinMaxPriorityQueue<Integer> inQueue = MinMaxPriorityQueue.<Integer>orderedBy(Comparator.reverseOrder())
                .create(Arrays.asList(5, 2, 7, 2, 4));

        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        tester.serialize(outputStream, inQueue);

        ByteArrayInputStream inputStream = new ByteArrayInputStream(outputStream.toByteArray());
        @SuppressWarnings("unchecked")
        MinMaxPriorityQueue<Integer> outQueue = (MinMaxPriorityQueue<Integer>) tester.deserialize(inputStream);

        System.out.println(inQueue);
        System.out.println(outQueue);
    }

    private final Kryo kryo;

    public MinMaxPriorityQueueSerializerTester() {
        this.kryo = new Kryo();
        registerMinMaxSerializer();
        allowForClassesWithoutNoArgConstructor(); // needed to serialize Ordering
    }

    private void registerMinMaxSerializer() {
        kryo.addDefaultSerializer(MinMaxPriorityQueue.class, new MinMaxPriorityQueueSerializer());
    }

    private void allowForClassesWithoutNoArgConstructor() {
        ((Kryo.DefaultInstantiatorStrategy) kryo.getInstantiatorStrategy())
                .setFallbackInstantiatorStrategy(new StdInstantiatorStrategy());
    }


    public void serialize(OutputStream out, MinMaxPriorityQueue<?> queue) {
        try (Output output = new Output(out)) {
            kryo.writeObject(output, queue);
        }
    }

    public MinMaxPriorityQueue<?> deserialize(InputStream in) {
        try (Input input = new Input(in)) {
            return kryo.readObject(input, MinMaxPriorityQueue.class);
        }
    }
}

【讨论】：

谢谢，我试过了，没有运气:-(
@elbaulp 我在 Java 中测试了这段代码，它工作正常。您是否在 Scala（Flink 之外）中测试过序列化？如果没有，请这样做（就像我在MinMaxPriorityQueueSerializerTester 中所做的那样），然后我们会知道更多。

【解决方案2】：

我最终放弃并尝试使用不同的数据结构并使用java.io.Serializable 使其可序列化。

这个数据结构是一个IntervalHeap implemented here，我刚刚在my project 中将它设为可序列化。

现在一切正常。

【讨论】：