Apache Flink中的不可序列化对象

Question

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

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

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

目前，我正在使用静态块来初始化其他类中的对象，使用前提条件。在主文件中选中NOTNULL（mgGenerator.mgGenerator），如果在接收器的平面图中使用它，它就会工作。

有没有办法创建一个不可序列化的依赖对象，它可能需要超过10秒才能在Flink的平面图或接收器中初始化？

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));
      }

    });

  }
}

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

Answer 1

使用打开通常是加载外部查找源的正确位置。超时有点奇怪，可能有一个配置围绕它。

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

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

我看不到使用代理序列化器模式的好处。它不必要地复杂（Java中的自定义序列化），并且没有什么好处。

Answer 2

在Open方法中执行此操作是100%正确的方法。Flink是为您提供超时异常还是对象？

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