Search code examples
scalaapache-sparkavro

Getting ClassCastException while trying to save file in avro format in spark

I am trying to process a file and then saving it in avro file format using saveAsNewAPIHadoopFile method in spark. Below is my program:

case class TrafficSchema(a: String, b: Int, c: Int, d: Int, e: Float)

 def main(args: Array[String]) {
val tableName: String = "CHICAGO_TRAFFIC_TRACKER"
System.setProperty("hadoop.home.dir", "D:\\")
val input_Path = "E:\\SharedVM\\Chicago_Traffic_Tracker_-_Historical_Congestion_Estimates_by_Region_-_2013-2018.csv"
val job = Job.getInstance
val schema = Schema.create(Schema.Type.STRING)
AvroJob.setOutputKeySchema(job, schema)
val avroOutputPath = "D:\\ScalaIDe_latest\\ChicagoTratfficTracker\\output\\Chicago_Traffic_Tracker-2013-2018_AVRO"
val conf = new SparkConf().setMaster("local").setAppName("TrafficTracker").set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  .set("spark.kryo.registrator", "com.jesperdj.example.MyRegistrator")
val sc = new SparkContext(conf)

val data = sc.textFile(input_Path)

val header = data.first()
val trackerData = data.filter(head => head != header).map(row => row.toString().trim().split(","))
  .map(f => {
    //println(f(0).toString()+"-"+f(1).toString().toInt+"-"+ f(2).toString().toInt+"-"+ f(3).toString().toInt+"-"+ f(4).toString())
    TrafficSchema(f(0).toString(), f(1).toString().toInt, f(2).toString().toInt, f(3).toString().toInt, f(4).toString().toFloat)
  })

val intermediateRDD = trackerData.mapPartitions(
  f = (iter: Iterator[TrafficSchema]) => iter.map(new AvroKey(_) -> NullWritable.get()))
intermediateRDD.saveAsNewAPIHadoopFile(
  avroOutputPath,
  classOf[AvroKey[TrafficSchema]],
  classOf[NullWritable],
  classOf[AvroKeyOutputFormat[GenericRecord]],
      job.getConfiguration)
  }

And below is the error I am getting:

org.apache.avro.file.DataFileWriter$AppendWriteException: java.lang.ClassCastException: com.jesperdj.example.TrafficSchema cannot be cast to java.lang.CharSequence
at org.apache.avro.file.DataFileWriter.append(DataFileWriter.java:308)
at org.apache.avro.mapreduce.AvroKeyRecordWriter.write(AvroKeyRecordWriter.java:77)
at org.apache.avro.mapreduce.AvroKeyRecordWriter.write(AvroKeyRecordWriter.java:39)
at 

org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12$$anonfun$apply$4.apply$mcV$sp(PairRDDFunctions.scala:1108)
    at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12$$anonfun$apply$4.apply(PairRDDFunctions.scala:1106)
    at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12$$anonfun$apply$4.apply(PairRDDFunctions.scala:1106)
    at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1277)
    at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12.apply(PairRDDFunctions.scala:1114)
    at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12.apply(PairRDDFunctions.scala:1085)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)
    at org.apache.spark.scheduler.Task.run(Task.scala:86)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:254)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
    at java.lang.Thread.run(Unknown Source)
Caused by: java.lang.ClassCastException: com.jesperdj.example.TrafficSchema cannot be cast to java.lang.CharSequence
    at org.apache.avro.generic.GenericDatumWriter.writeString(GenericDatumWriter.java:267)
    at org.apache.avro.specific.SpecificDatumWriter.writeString(SpecificDatumWriter.java:71)
    at org.apache.avro.generic.GenericDatumWriter.writeWithoutConversion(GenericDatumWriter.java:128)
    at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:75)
    at org.apache.avro.reflect.ReflectDatumWriter.write(ReflectDatumWriter.java:159)
    at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:62)
    at org.apache.avro.file.DataFileWriter.append(DataFileWriter.java:302)
    ... 14 more

I am not sure what should i defined in schema.create method. I tried with record but I am not sure how can i define traffic schema in the Schema.create method

Sample CSV Data :

CSV Data

Solution

  • Indeed, you have to define an avro schema, or use external librairies such as avro4s, to get it from your case class.

    Using native Avro:

    val schema = "{\"type\":\"record\",\"name\":\"TrafficSchema\",\"namespace\":\"your.project.namespace\",\"fields\":[{\"name\":\"str\",\"type\":\"string\"},{\"name\":\"i\",\"type\":\"int\"},{\"name\":\"i1\",\"type\":\"int\"},{\"name\":\"i2\",\"type\":\"int\"},{\"name\":\"fl\",\"type\":\"float\"}]}"

val trafficSchema = new Schema.Parser().parse(schema)

    Using avro4s:

    import com.sksamuel.avro4s.AvroSchema

val trafficSchema: Schema = AvroSchema[TrafficSchema]