ERROR OneForOneStrategy

如果你在 Spark Streaming 里启用 checkpointing，forEachRDD 函数使用的对象都应该可以被序列化(Serializable)。否则会出现这样的异常 “ERROR OneForOneStrategy: … java.io.NotSerializableException:”

  1. JavaStreamingContext jssc = new JavaStreamingContext(sc, INTERVAL);
  3. // This enables checkpointing.
  4. jssc.checkpoint("/tmp/checkpoint_test");
  6. JavaDStream<String> dStream = jssc.socketTextStream("localhost", 9999);
  8. NotSerializable notSerializable = new NotSerializable();
  9. dStream.foreachRDD(rdd -> {
  10.       if (rdd.count() == 0) {
  11.         return null;
  12.       }
  13.       String first = rdd.first();
  15.       notSerializable.doSomething(first);
  16.       return null;
  17.     }
  18. );
  20. // This does not work!!!!

按照下面的方式之一进行修改，上面的代码才能正常运行：

  • 在配置文件里面删除 jssc.checkpoint 这一行关闭 checkpointing。
  • 让对象能被序列化。
  • 在 forEachRDD 函数里面声明 NotSerializable，下面的示例代码是可以正常运行的：
  1. JavaStreamingContext jssc = new JavaStreamingContext(sc, INTERVAL);
  3. jssc.checkpoint("/tmp/checkpoint_test");
  5. JavaDStream<String> dStream = jssc.socketTextStream("localhost", 9999);
  7. dStream.foreachRDD(rdd -> {
  8.       if (rdd.count() == 0) {
  9.         return null;
  10.       }
  11.       String first = rdd.first();
  12.       NotSerializable notSerializable = new NotSerializable();
  13.       notSerializable.doSomething(first);
  14.       return null;
  15.     }
  16. );
  18. // This code snippet is fine since the NotSerializable object
  19. // is declared and only used within the forEachRDD function.

阅读原文