Apache Flink 中的不可序列化对象答案

【问题标题】：Non Serializable object in Apache FlinkApache Flink 中的不可序列化对象
【发布时间】：2020-04-29 23:30:08
【问题描述】：

我正在使用 Apache Flink 对流数据执行分析。

我正在使用一个依赖项，其对象的创建时间超过 10 秒，因为它在初始化之前会读取 hdfs 中存在的多个文件。

如果我在 open 方法中初始化对象，我会得到一个超时异常，如果在 sink/flatmap 的构造函数中，我会得到序列化异常。

目前我正在使用静态块在其他类中初始化对象，在主文件中使用 Preconditions.checkNotNull(MGenerator.mGenerator)，然后如果在接收器的平面图中使用它就可以工作。

有没有办法在 Flink 的 flatmap 或 sink 中创建一个可能需要超过 10 秒才能初始化的不可序列化依赖对象？

public class DependencyWrap {

  static MGenerator mGenerator;

  static {
    final String configStr = "{}";
    final Config config = new Gson().fromJson(config, Config.class);
    mGenerator = new MGenerator(config);
  }

}

public class MyStreaming {

  public static void main(String[] args) throws Exception {

    Preconditions.checkNotNull(MGenerator.mGenerator);
    final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(parallelism);
    ...
    input.flatMap(new RichFlatMapFunction<Map<String,Object>,List<String>>() {

      @Override
      public void open(Configuration parameters) {
      }

      @Override
      public void flatMap(Map<String,Object> value, Collector<List<String>> out) throws Exception {

        out.collect(MFVGenerator.mfvGenerator.generateMyResult(value.f0, value.f1));
      }

    });

  }
}

另外，如果我对这个问题有错误，请纠正我。

【问题讨论】：

能分享一下异常的堆栈跟踪吗？我会遇到类似的问题（open() 的长时间延迟导致超时），堆栈跟踪帮助我确定了要增加哪个 Flink 配置值。

标签： apache-flink flink-streaming

【解决方案1】：

在 Open 方法中执行此操作是 100% 正确的方法。 Flink 给你的是超时异常，还是对象？

作为最后的方法，您可以将对象包装在一个包含对象及其 JSON 字符串或 Config（Config 可序列化？）的类中，对象标记为瞬态，然后覆盖 ReadObject/WriteObject 方法以调用构造函数。如果 mGenerator 对象本身是无状态的（如果不是，您将遇到其他问题），则序列化代码应该只在作业分发给任务管理器时调用一次。

【讨论】：

【解决方案2】：

使用open 通常是加载外部查找源的正确位置。超时有点奇怪，说不定周围有配置呢。

但是，如果使用静态加载器（如您所做的静态类或单例）非常庞大，则您只需为同一任务管理器上的所有并行任务实例加载一次即可。因此，您可以节省内存和 CPU 时间。这对您来说尤其如此，因为您在两个单独的任务中使用相同的数据结构。此外，静态加载器可以在第一次使用时延迟初始化，以避免open超时。

这种方法的明显缺点是代码的可测试性会受到影响。有一些方法可以解决这个问题，如果有兴趣，我可以扩展。

我没有看到使用代理序列化程序模式的好处。它过于复杂（Java 中的自定义序列化）并且几乎没有什么好处。

【讨论】：