当第一行是模式时，如何从Spark中的csv（使用scala）创建数据帧？

Question

我是Spark的新手，我正在使用scala编程。我想从HDFS或S3中读取一个文件，并将其转换为Spark数据帧。Csv文件的第一行是模式。但是，如何创建具有未知列的模式的数据框架呢？我使用下面的代码为一个已知的模式创建数据框架。

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;

}

Answer 1

亲爱的哈米德，你可以试试下面的代码

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
    }
  }

希望它能帮助你：）