java - 无法转换为 org.apache.spark.serializer.Serializer

Question

我正在尝试使用 Java 解决 HashMap 的 Spark 序列化问题。我指的是链接 Save Spark Dataframe into Elasticsearch - Can’t handle type exception .

现在我遇到了以下问题:

java.lang.ClassCastException: com.spark.util.umf.MyKryoRegistrator cannot be cast to org.apache.spark.serializer.Serializer    at org.apache.spark.SparkEnv$.create(SparkEnv.scala:259)    at org.apache.spark.SparkEnv$.createDriverEnv(SparkEnv.scala:163)    at org.apache.spark.SparkContext.createSparkEnv(SparkContext.scala:267)    at org.apache.spark.SparkContext.(SparkContext.scala:270)    at org.apache.spark.api.java.JavaSparkContext.JavaSparkContext.scala:61)    at com.spark.util.umf.MyMain.main(MyMain.java:46)    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)    at java.lang.reflect.Method.invoke(Method.java:606)    at org.apache.spark.deploy.yarn.ApplicationMaster$$anon$2.run(ApplicationMaster.scala:480)15/10/16 01:47:22 INFO yarn.ApplicationMaster: Final app status:FAILED, exitCode: 15, (reason: User class threw exception:com.spark.util.umf.MyKryoRegistrator cannot be cast toorg.apache.spark.serializer.Serializer)

I create my Kryo registrator as followed :

import java.io.Serializable;
import org.apache.spark.serializer.KryoRegistrator;
import com.esotericsoftware.kryo.Kryo;

public class MyKryoRegistrator implements KryoRegistrator, Serializable {
    @Override
    public void registerClasses(Kryo kryo) {
        // Product POJO associated to a product Row from the DataFrame            
        kryo.register(MyRecord.class); 
    }
}

主要方法:

public static void main(String args[]){

    SparkConf sConf= new SparkConf().setAppName("SparkTestJob");
    sConf.set( "spark.driver.allowMultipleContexts", "true");
    //Kryo kryo = new Kryo();;
    //kryo.setDefaultSerializer(MyRecord.class);
    //my.registerClasses(kryo);
    sConf.set("spark.serializer","com.spark.util.umf.MyKryoRegistrator");

    [...]
}

Answer 1

根据我在您的问题中提到的链接中提供的答案，您可以看到我已经定义了两个参数:

spark.serializer 和 spark.kryo.registrator

所以你必须设置这两个参数。

如果你设置注册器而不设置序列化器，kryo序列化器将不会被设置。