1

I am trying to solve a Spark serialization issue with HashMaps using Java. I am referring to the link Save Spark Dataframe into Elasticsearch - Can’t handle type exception .

Now I am hitting the following issue: 

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 to
org.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); 
    }
}

Main method :

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

    [...]
}
eliasah
  • 39,588
  • 11
  • 124
  • 154
Harisyam
  • 107
  • 3
  • 15
  • Where is `main` defined? Please post full code, as it looks like `MyKryoRegistrator` is an inner class, if so try `public static class MyKryoRegistrator` ... – earcam Oct 16 '15 at 02:16
  • Ok . It was my mistake . I used like "sConf.set("spark.serializer","com.spark.util.umf.MyKryoRegistrator");" It should be like sConf.set("spark.kryo.registrator","com.spark.util.umf.MyKryoRegistrator"); – Harisyam Oct 16 '15 at 02:21
  • you can put the comment as an answer to your own question ;-) – earcam Oct 16 '15 at 02:26

1 Answers1

3

Based of the answer I provided in the link that you have mentioned in your question, you can see that you I have defined both parameters :

spark.serializer and spark.kryo.registrator

So you have to set both parameters.

If you set the registrator without setting the serializer, the kryo serializer won't be set.

eliasah
  • 39,588
  • 11
  • 124
  • 154