The Avro schema is defined using JSON. In our case, we'll just use the same Student.csv as the input file. So, let's code the four fields— id, name, phone, and email—in the schema:

{"namespace": "studentavro.avro",
 "type": "record",
 "name": "StudentAvro",
 "fields": [
     {"name": "id", "type": ["string", "null"]},
     {"name": "name",  "type": ["string", "null"]},
     {"name": "phone", "type": ["string", "null"]},
     {"name": "email", "type": ["string", "null"]}
 ]
}

Probably, you are already familiar with Avro, or you have already understood the schema just by taking a look at it, but let me bore you with some explanation of the schema anyway.

The namespace and name attributes in the JSON translate into our package name and class name in our world, respectively. So, our generated class will have a fully qualified name as studentavro.avro.StudentAvro. The "record" (of the type attribute) is one of the complex types in Avro (http://avro.apache.org/docs/1.7.6/spec.html#schema_complex). Let me rephrase this again. A record roughly translates to classes in Java/Scala. It is at the topmost level in the schema hierarchy. A record can have multiple fields encapsulated inside it, and these fields can be primitives (https://avro.apache.org/docs/1.7.7/spec.html#schema_primitive) or other complex types. The last bit about the type having an array of types is interesting ("type": ["string", "null"]). It just means that the field can be more than one type. In Avro terms, it is called a union.

Now that we are done with the schema, let's save this file with an extension of .avsc. I have saved it as student.avsc in the src/main/avro directory.

The next step is to generate a class from the schema. The reason we stored the avro schema file in the src/main/avro folder is this: we'll be using an sbt-avro plugin (https://github.com/cavorite/sbt-avro) to generate a Java class from the schema. Configuring the plugin is as easy as configuring any other plugin for SBT:

This step is very similar to the previous recipe in the sense that we use the convertCSVToStudents function to generate an RDD of the StudentAvro object. Also, since this isn't a Scala class and the generated Java object comes up with a builder inside it, we use the builder to construct the class fluently (http://en.wikipedia.org/wiki/Fluent_interface):

val conf = new SparkConf().setAppName("AvroModelToParquet").setMaster("local[2]")

  val sc = new SparkContext(conf)
  val sqlContext = new SQLContext(sc)
  sqlContext.setConf("spark.sql.parquet.binaryAsString", "true")
  val rddOfStudents = convertCSVToStudents("StudentData.csv", sc)


//The CSV has a header row.  Zipping with index and skipping the first row
  def convertCSVToStudents(filePath: String, sc: SparkContext): RDD[StudentAvro] = {
    val rddOfStudents: RDD[StudentAvro]=sc.textFile(filePath).flatMap(eachLine => {
      val data = eachLine.split("\|")
      if (data(0) == "id") None
      else Some(StudentAvro.newBuilder()
        .setId(data(0))
        .setName(data(1))
        .setPhone(data(2))
        .setEmail(data(3)).build())
    })
    rddOfStudents
  }

This is a tricky step and involves multiple substeps. Let's decipher this step backwards. We fall back to RDD[StudentAvro] in this example instead of a DataFrame because DataFrames can be constructed only from an RDD of case classes (or classes that extend Product, as we saw earlier in this chapter) or from RDD[org.apache.spark.sql.Row]. If you prefer to use DataFrames, you can read the CSV as an array of values, and use RowFactory.create for each array of values. Once an RDD[Row] is available, we can use sqlContext.createDataFrame to convert it to a DataFrame:

As always, let's read the file back for confirmation. The function to be called for this is newAPIHadoopFile, which accepts a similar set of parameters as saveAsNewAPIHadoopFile: the name of the file, InputFormat, the key class, the value class, and finally the job configuration. Note that we are using newAPIHadoopFile instead of the previously used the parquetFile method. This is because we are reading from a Hadoop sequence file:

//Reading the file back for confirmation.
  ParquetInputFormat.setReadSupportClass(job, classOf[AvroWriteSupport])
  val readStudentsPair = sc.newAPIHadoopFile("studentAvroPq", classOf[AvroParquetInputFormat[StudentAvro]], classOf[Void], classOf[StudentAvro], job.getConfiguration())
  val justStudentRDD: RDD[StudentAvro] = readStudentsPair.map(_._2)
  val studentsAsString = justStudentRDD.collect().take(5).mkString("
")
  println(studentsAsString)

This is the output:

We'll also use Parquet tools to confirm that the schema that is stored in the Parquet file is indeed an avro schema:

/Users/Gabriel/Dropbox/arun/ScalaDataAnalysis/git/parquet-mr/parquet-tools/target/parquet-tools-1.6.0rc3/parquet-tools meta /Users/Gabriel/Dropbox/arun/ScalaDataAnalysis/Code/scaladataanalysisCB-tower/chapter3-data-loading-parquet/studentAvroPq

Yup! Looks like it is! The extra section in meta does confirm that the avro schema is stored:

creator:     parquet-mr
extra:       parquet.avro.schema = {"type":"record","name":"StudentAvro","namespace":"studentavro.avro","fields":[{"name":"id","type":[{"type":"string","avro.java.string":"Stri [more]...