如何在 Flink 中调试可序列化异常?

问题描述

我遇到了几个可序列化的异常，我在 Flink 的互联网和文档上做了一些搜索；有一些著名的解决方案，如瞬态、扩展可序列化等.每次异常的来源都非常清楚，但在我的情况下，我无法找到它没有被序列化的确切位置.

I've encountered several serializable exceptions, and I did some searching on Flink's internet and doc; there are some famous solutions like transient, extends Serializable etc. Each time the origin of exception is very clear, but in my case, i am unable to find where exactly it is not serialized.

问:我应该如何调试这种异常?

A.scala:

class executor ( val sink: SinkFunction[List[String]] {
    def exe(): Unit = {
        xxx.....addSink(sinks)
    }
}

B.scala:

class Main extends App {
  def createSink: SinkFunction[List[String]] = new StringSink()

  object StringSink {
    // static
    val stringList: List[String] = List()
  }

  // create a testing sink
  class StringSink extends SinkFunction[List[String]] {
    override def invoke(strs: List[String]): Unit = {
        // add strs into the variable "stringList" of the compagin object StringSink
    }
  }

  new executor(createSink()).exe()

  // then do somethings with the strings
}

例外是:

SinkFunction 的实现是不可序列化的.这对象可能包含或引用不可序列化的字段.

我发现的两个可疑点:

1. StringSink 的实例被传递到另一个文件中.
2. 在StringSink的类中，它使用了一个静态变量stringList其组合对象.

1. The instance of StringSink is passed into another file.
2. In the class of StringSink, it uses a static variable stringListof its compagin object.

推荐答案

我遇到了类似的问题.过去需要很长时间才能找出哪些成员/对象不可序列化.异常日志并没有真正的帮助.

I faced similar problems. It used to take longtime to find out what member/object is not serializable. The exception logs are not really helpful.

帮助我的是以下 JVM 选项，它可以在异常跟踪中启用更多详细信息.

What helped me is the following JVM option, which enables more details in exception trace.

启用此选项...

-Dsun.io.serialization.extendedDebugInfo=true

-Dsun.io.serialization.extendedDebugInfo=true

这篇关于如何在 Flink 中调试可序列化异常?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！