We can actually create a DataFrame directly from CSV using the com.databricks/spark-csv file, as we saw in Chapter 2, Getting Started with Apache Spark DataFrames, but for this recipe, we'll just tokenize the CSV and create classes from it. The input CSV has a header column. So, the conversion process involves skipping the first row.

There are just two interesting things that you might notice in the code:

sqlContext.setConf("spark.sql.parquet.binaryAsString","true")

Some Parquet producing systems, such as Impala, binary encode the strings. In order to work around this issue, we set the following configuration, which says that if it sees binary data, it should be treated as a string:

Instead of using sqlContext.createDataFrame, we just use a toDF() on the RDD[Student]. The SQLContext.Implicits object has a number of implicit conversions that help us convert an RDD[T] to a DataFrame directly. The only requirement for us, as expected, is to import the implicits:

  import sqlContext.implicits._

The rest of the code is the same as we saw earlier:

val conf = new SparkConf().setAppName("CaseClassToParquet").setMaster("local[2]")
  val sc = new SparkContext(conf)
  val sqlContext = new SQLContext(sc)

  //Treat binary encoded values as Strings
  sqlContext.setConf("spark.sql.parquet.binaryAsString","true")

  import sqlContext.implicits._

  //Convert each line into Student
  val rddOfStudents = convertCSVToStudents("StudentData.csv", sc)

  //Convert RDD[Student] to a Dataframe using sqlContext.implicits
  val studentDFrame = rddOfStudents.toDF()

The convertCSVToStudents method, which converts each line into a Student object, looks like this:

def convertCSVToStudents(filePath: String, sc: SparkContext): RDD[Student] = {
    val rddOfStudents: RDD[Student] =sc.textFile(filePath).flatMap(line => {
      val data = line.split("\|")
      if (data(0) == "id") None else Some(Student(data(0), data(1), data(2), data(3)))
    })
    rddOfStudents
  }

This is just a one-liner once we have the DataFrame. This can be done using either the saveAsParquetFile or the save method. If you wish to save it in a Hive table (https://hive.apache.org/), then there is also a saveAsTable method for you:

//Save DataFrame as Parquet using saveAsParquetFile
studentDFrame.saveAsParquetFile("studentPq.parquet")

//OR

//Save DataFrame as Parquet using the save method
studentDFrame.save("studentPq.parquet", "parquet", SaveMode.Overwrite)

The save methods create a directory in the location that you specify (here, we simply store it in our project directory). The directory holds the files that represent the serialized data. It is not entirely human readable, but you may notice that the data of a single column is stored together.

Just as we do for the rest of the recipes, let's read the file and sample the data for confirmation:

//Read data for confirmation
  val pqDFrame=sqlContext.parquetFile("studentPq.parquet")
  pqDFrame.show()

The following is the output:

Other than using the printSchema method of the DataFrame to inspect the schema, we can use some interesting parquet tools provided as part of the parquet project to get a variety of other information.

Since Spark 1.4.1 uses Parquet 1.6.0rc3, we'll need to download that version of the tools from the Maven repository. The executables and the JARs can be downloaded as one bundle from https://repo1.maven.org/maven2/com/twitter/parquet-tools/1.6.0rc3/parquet-tools-1.6.0rc3-bin.tar.gz.

Let's put the tools into action. Specifically, we'll do three things in this step:

The preceding command will display only two rows because of the additional -n 2 parameter.

The following is the output of this command:

id = 1
name = Burke
phone = 1-300-746-8446
email = [email protected]

id = 2
name = Kamal
phone = 1-668-571-5046
email = [email protected]

Optionally, if you wish to display all the records in the file, you can use the cat parameter with the parquet-tools command:

parquet-tools cat part-r-00001.parquet

As you can see from the meta information, the data is gzipped by default. In order to use Snappy compression, all that we need to do is set a configuration to our SQLContext (actually the SQLConf of SQLContext). There's just one catch with regard to enabling Lempel–Ziv–Oberhumer (LZO) compression—we are required to install native-lzo on all the machines where this data is stored. Otherwise, we get a "native-lzo library not available" error message.

Let's enable Snappy (http://google.github.io/snappy/) compression by passing the configuration parameter of Parquet compression to Snappy:

sqlContext.setConf("spark.sql.parquet.compression.codec", "snappy")

After running the program, let's use the parquet-tools meta command to verify it:

parquet-tools meta part-r-00000-aee54b77-288e-44b2-8f36-53b38a489e8d.snappy.parquet