How to create Data frame from csv in Spark(using scala) when the first line is the schema?

Question

I am new to Spark and I am coding using scala. I want to read a file from HDFS or S3 and convert it into Spark Data frame. The first line of the Csv file is the schema. but how can I create a dataframe with a schema having unknown columns? I was using the following piece of code to create the dataframe for a known schema.

def loadData(path:String): DataFrame = {

  val rdd = sc.textFile(path);
  val firstLine = rdd.first();
  val schema = StructType(firstLine.split(',').map(fieldName=>StructField(fieldName,StringType,true)));

  val noHeader = rdd.mapPartitionsWithIndex( 
    (i, iterator) => 
      if (i == 0 && iterator.hasNext) { 
         iterator.next 
         iterator 
        } else iterator)


  val rowRDD = noHeader.map(_.split(",")).map(p => Row(p(0), p(1), p(2), p(3), p(4),p(5)))

  val dataFrame = sqlContext.createDataFrame(rowRDD, schema);
  return dataFrame;

}

You can try to use Spark CSV Library of Databricks: https://github.com/databricks/spark-csv — Leonard, Jul 14 '15 at 05:15
possible duplicate of [How to read csv file as DataFrame?](http://stackoverflow.com/questions/29704333/how-to-read-csv-file-as-dataframe) The spark-csv library allows you to say whether there's a header line or not — The Archetypal Paul, Jul 14 '15 at 07:06

Hafiz Mujadid · Accepted Answer · 2016-06-20T16:09:48.113

You can try following code dear hammad

val sc = new SparkContext(new SparkConf().setMaster("local").setAppName("test"))
val sqlcon = new SQLContext(sc)
//comma separated list of columnName:type

def main(args:Array[String]){
var schemaString ="Id:int,FirstName:text,LastName:text,Email:string,Country:text"
val schema =
      StructType(
        schemaString.split(",").map(fieldName => StructField(fieldName.split(":")(0),
          getFieldTypeInSchema(fieldName.split(":")(1)), true)))
val rdd=sc.textFile("/users.csv")
val noHeader = rdd.mapPartitionsWithIndex( 
(i, iterator) => 
  if (i == 0 && iterator.hasNext) { 
     iterator.next 
     iterator 
    } else iterator)
 val rowRDDx =noHeader.map(p => {
      var list: collection.mutable.Seq[Any] = collection.mutable.Seq.empty[Any]
      var index = 0
      var tokens = p.split(",")
      tokens.foreach(value => {
        var valType = schema.fields(index).dataType
        var returnVal: Any = null
        valType match {
          case IntegerType => returnVal = value.toString.toInt
          case DoubleType => returnVal = value.toString.toDouble
          case LongType => returnVal = value.toString.toLong
          case FloatType => returnVal = value.toString.toFloat
          case ByteType => returnVal = value.toString.toByte
          case StringType => returnVal = value.toString
          case TimestampType => returnVal = value.toString
        }
        list = list :+ returnVal
        index += 1
      })
      Row.fromSeq(list)
    })
val df = sqlcon.applySchema(rowRDDx, schema)
}
def getFieldTypeInSchema(ftype: String): DataType = {

    ftype match {
      case "int" => return IntegerType
      case "double" => return DoubleType
      case "long" => return LongType
      case "float" => return FloatType
      case "byte" => return ByteType
      case "string" => return StringType
      case "date" => return TimestampType
      case "timestamp" => return StringType
      case "uuid" => return StringType
      case "decimal" => return DoubleType
      case "boolean" => BooleanType
      case "counter" => IntegerType
      case "bigint" => IntegerType
      case "text" => return StringType
      case "ascii" => return StringType
      case "varchar" => return StringType
      case "varint" => return IntegerType
      case default => return StringType
    }
  }

Hope it will help u. :)

How to create Data frame from csv in Spark(using scala) when the first line is the schema?

1 Answers1