Merge pull request #58 from badrinathpatchikolla/master

Added CSV Deserializer

src/main/scala/com/github/music/of/the/ainur/almaren/builder/core/Deserializer.scala

@@ -15,6 +15,8 @@ private[almaren] trait Deserializer extends Core {
       XMLDeserializer(columnName,schemaInfo,options,autoFlatten)
     def avro: State =
       AvroDeserializer(columnName,None,options,autoFlatten,schemaInfo.getOrElse(throw SchemaRequired(decoder)))
+    def csv: State =
+      CSVDeserializer(columnName, schemaInfo, options, autoFlatten)
 
     decoder.toUpperCase match {
       case "JSON" => json

src/main/scala/com/github/music/of/the/ainur/almaren/state/core/Deserializer.scala

@@ -101,3 +101,23 @@ case class XMLDeserializer(columnName: String, schema: Option[String], options:
       .withColumn(columnName, from_xml(col(columnName), xmlSchema, options))
   }
 }
+
+case class CSVDeserializer(columnName: String, schema: Option[String], options: Map[String, String], autoFlatten: Boolean) extends Deserializer {
+
+  import org.apache.spark.sql.functions._
+  import collection.JavaConversions._
+
+  override def deserializer(df: DataFrame): DataFrame = {
+    import df.sparkSession.implicits._
+    logger.info(s"columnName:{$columnName}, schema:{$schema}, options:{$options}, autoFlatten:{$autoFlatten}")
+    df.withColumn(columnName,
+      from_csv(
+        col(columnName),
+        schema.getOrElse(getSchemaDDL(df.selectExpr(columnName).as[(String)])),
+        options
+      ))
+  }
+
+  private def getSchemaDDL(df: Dataset[String]): String =
+    getDDL(getReadWithOptions.csv(sampleData(df)))
+}

src/test/scala/com/github/music/of/the/ainur/almaren/Test.scala

@@ -112,6 +112,7 @@ class Test extends FunSuite with BeforeAndAfter {
   deserializerJsonTest()
   deserializerXmlTest()
   deserializerAvroTest()
+  deserializerCsvTest()
   testInferSchemaJsonColumn()
   testInferSchemaDataframe(moviesDf)
 
@@ -464,6 +465,27 @@ class Test extends FunSuite with BeforeAndAfter {
     test(jsonschmeadf, resDf, "Deserialize JSON Schema")
   }
 
+  def deserializerCsvTest(): Unit = {
+    val df = Seq(
+      ("John,Chris", "Smith", "London"),
+      ("David,Michael", "Jones", "India"),
+      ("Joseph,Mike", "Lee", "Russia"),
+      ("Chris,Tony", "Brown", "Indonesia"),
+    ).toDF("first_name", "last_name", "country")
+    val newCsvDF = almaren.builder
+      .sourceDataFrame(df)
+      .deserializer("CSV", "first_name", options = Map("header" -> "false"))
+      .batch
+    val newCsvSchemaDf = almaren.builder
+      .sourceDataFrame(df)
+      .deserializer("CSV", "first_name", Some("`first_name_1` STRING,`first_name_2` STRING"), Map("header" -> "true"))
+      .batch
+    val csvDf = spark.read.parquet("src/test/resources/data/csvDeserializer.parquet")
+    val csvSchemaDf = spark.read.parquet("src/test/resources/data/csvDeserializerSchema.parquet")
+    test(newCsvDF, csvDf, "Deserialize CSV")
+    test(newCsvSchemaDf, csvSchemaDf, "Deserialize CSV Schema")
+  }
+
   def deserializerXmlTest(): Unit = {
     val xmlStr = Seq(
       """ <json_string>