Support Questions
Find answers, ask questions, and share your expertise
Announcements
Alert: Welcome to the Unified Cloudera Community. Former HCC members be sure to read and learn how to activate your account here. Want to know more about what has changed? Check out the Community News blog.

Spark integration with Avro(In CDH 5.1.0)

Spark integration with Avro(In CDH 5.1.0)

Explorer

Hi All, How to load Avro message to Spark? I have avro stream message that I want to load in Spark and do groupBykey operation as per my requirement. My work:- sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer"); sparkConf.set("spark.kryo.registrator", "xyz.PlatformKryoRegistrator"); public class PlatformKryoRegistrator implements KryoRegistrator { public void registerClasses(Kryo kryo) { kryo.register(ResourceMessage.class, new FieldSerializer(kryo, ResourceMessage.class)); } } ///ResourceMessage class gets generated from avro tools public class ResourceMessage extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord { ......... } Problem is that it is not able to register custom PlatformKryoRegistrator class.I am not sure what I am doing wrong. Please suggest me. Thanks Nishi

1 REPLY 1

Re: Spark integration with Avro(In CDH 5.1.0)

Explorer

Hi All,

How to load Avro message to Spark? I have avro stream message that I want to load in Spark and do groupBykey operation as per my requirement.

 

My work:-

 

sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");

sparkConf.set("spark.kryo.registrator", "xyz.PlatformKryoRegistrator");

 

public class PlatformKryoRegistrator implements KryoRegistrator {

public void registerClasses(Kryo kryo) {

  kryo.register(ResourceMessage.class, new FieldSerializer(kryo, ResourceMessage.class)); }

}

 

///ResourceMessage class gets generated from avro tools

public class ResourceMessage extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord { ......... }

 

Problem is that it is not able to register custom PlatformKryoRegistrator class.I am not sure what I am doing wrong.