使用Java实现Apache Spark对象序列化

Question

我正在尝试准备一个库（用Java编写）来运行在Apache-Spark上。由于该库有数百个类，并且仍处于活跃的开发阶段，所以我不想一一序列化所有的类。相反，我搜索了另一个方法，并找到了这个方法，但它同样不能解决序列化问题。

下面是代码示例：

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

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

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 + "]";
}}

这会产生年份4d的“对象不可序列化”异常：

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

顺便说一下,如果我将命令Action collect（）替换为foreach（func）,

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

Answer 1

解决方案1（您不会使用它，因为通过使类实现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);
    }
}

建议类包含一个无参数的构造函数。

如果您对Kyro默认提供的seriliazers不满意，或者您有复杂的类，您可以始终定义自己的类。