transpose.scala

import spark.implicits._
import org.apache.spark.sql.types._
import org.apache.spark.sql.Row

val schemaString = sc.textFile("/data.csv").take(1)(0)
val rdd = sc.textFile("/data.csv").filter(line => line != schemaString)
val fields = (schemaString.split(",").slice(0, 9) ++ Array("stat", "value"))
  .map(fieldName => {
    if (fieldName contains "value")
      StructField(fieldName, DoubleType, nullable = true)
    else
      StructField(fieldName, StringType, nullable = true)
  })
val schema = StructType(fields)

val rowRDD = rdd.flatMap(line => {
  val fields = line.split(",")
  val keyFields = fields.slice(0,9)
  fields.slice(10, fields.size).zipWithIndex.map(x => {
    Row(keyFields ++ Array("stat_type1" + (x._2).toString, x._1.toDouble):_*)
  })
})
  
val df = spark.createDataFrame(rowRDD, schema)
df.createOrReplaceTempView("base")