Object not serializable error on org.apache.avro.generic.GenericData$Record

Question

I am using the following code to create an RDD for a file I imported from MySQL with Sqoop into Hive:

def rddFromParquetHdfsFile(path: String): RDD[GenericRecord] = {
val job = new Job()
FileInputFormat.setInputPaths(job, path)
ParquetInputFormat.setReadSupportClass(job,
    classOf[AvroReadSupport[GenericRecord]])
return sc.newAPIHadoopRDD(job.getConfiguration,
    classOf[ParquetInputFormat[GenericRecord]],
    classOf[Void],
    classOf[GenericRecord]).map(x => x._2)
}
val warehouse = "hdfs://quickstart/user/hive/warehouse/"
val order_items = rddFromParquetHdfsFile(warehouse + "order_items");
val products = rddFromParquetHdfsFile(warehouse + "products");

I now try to view the first 5 products:

products.take(5)

and I end up with the following error:

org.apache.spark.SparkException: 
Job aborted due to stage failure: Task 0.0 in stage 0.0 (TID 0) had a 
not serializable result: org.apache.avro.generic.GenericData$Record
Serialization stack:
- object not serializable (class:   
  org.apache.avro.generic.GenericData$Record, value: {"product_id": 1,  
  "product_category_id": 2, "product_name": "Quest Q64 10 FT. x 10 FT. Slant  Leg Instant U", "product_description": "", "product_price": 59.98,  "product_image": "http://images.acmesports.sports /Quest+Q64+10+FT.+x+10+FT.+Slant+Leg+Instant+Up+Canopy"})
- element of array (index: 0)
- array (class [Lorg.apache.avro.generic.GenericRecord;, size 4)
at   org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1294)

Any suggestions about how to get around this ?

Answer 1

Try this with Spark conf:

conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") 
conf.registerKryoClasses(Array(classOf[org.apache.avro.generic.GenericData.Record]))