java - 使用 Java 的 Apache Spark 对象序列化

Question

我正在尝试准备一个库(用 Java 编写)以在 Apache-Spark 上运行。由于 Library 有数百个类，并且仍处于活跃的开发阶段，我不想将它们一一连载。相反，我搜索了另一种方法并找到了 this ，但它同样没有解决序列化问题。

这里是代码示例:

    List<Integer> data = Arrays.asList(1,2,3,4,5,6);
    JavaRDD<Integer> distData = sc.parallelize(data);
    JavaRDD<Year4D> years = distData.map(y -> func.call(y));
    List<Year4D> years1 = years.collect();

其中 func 是使用 Year4D 生成 4 位数字年份的函数；

    private static Function<Integer, Year4D> func = new Function<Integer, Year4D>() {
    public Year4D call(Integer arg0) throws Exception {
        return new Year4D(arg0);
    }};

并且 Year4D 没有实现 Serializable；

public class Year4D{
private int year = 0;
public Year4D(int year) {
    if (year < 1000) year += (year < 70) ? 2000 : 1900;
    this.year = year;
}
public String toString() {
    return "Year4D [year=" + year + "]";
}}

为 Year4D 产生“对象不可序列化”异常:

Job aborted due to stage failure: Task 6.0 in stage 0.0 (TID 6) had a not serializable result...

顺便说一句，如果我将 Command Action collect() 替换为 foreach(func) 它会起作用，

那么，我的问题是为什么 collect() 不起作用？

如果这种方法不好，那么处理包含大量复杂类的 Java 库的最佳实践是什么？

附言。@Tzach 说 Year4D 没有正确包装，所以实际上它没有序列化，那么正确的实现是什么？

Answer 1

解决方案 1(您不会使用它，因为通过使它们 implement Serializable 来修改每个类更容易):创建实现 的包装类>Serializable 并覆盖它们的 writeObject 和 readObject 方法

public class Year4DWraper implements Serializable{

    private Year4D year4d;

    public Year4DWraper(Year4D year4d) {
        this.year4d = year4d;
    }
    public Year4D getYear4D(){
        return yeard4D;
    }

    private void writeObject(ObjectOutputStream os)
            throws IOException {
       os.writeInt(year4D.getYear());

    }

    private void readObject(ObjectInputStream is)
            throws IOException, ClassNotFoundException {
       int year = is.readInt();
       year4D = new Yeard4D(year);
    }

}

解决方案 2:使用 Kyro为你做序列化/反序列化

SparkConf conf = new SparkConf();
conf.set("spark.kryo.registrator", "org.apache.spark.examples.MyRegistrator");
...

public class MyRegistrator implements KryoRegistrator {
    public void registerClasses(Kryo kryo) {
        kryo.register(Year4D.class);
    }
}

建议类包含 no-arg constructor .

By default, most classes will end up using FieldSerializer. It essentially does what hand written serialization would, but does it automatically. FieldSerializer does direct assignment to the object's fields. If the fields are public, protected, or default access (package private) and not marked as final, bytecode generation is used for maximum speed (see ReflectASM). For private fields, setAccessible and cached reflection is used, which is still quite fast.

如果您对 Kyro 默认提供的序列化器不满意或者您有复杂的类，您可以随时定义自己的。