Comment faire correspondre Dataframe noms de colonne à la Scala cas des attributs de classe?

Les noms de colonne dans cet exemple, à partir de l'étincelle-sql viennent de la case class Person.

case class Person(name: String, age: Int)

val people: RDD[Person] = ... //An RDD of case class objects, from the previous example.

//The RDD is implicitly converted to a SchemaRDD by createSchemaRDD, allowing it to be stored using Parquet.
people.saveAsParquetFile("people.parquet")

https://spark.apache.org/docs/1.1.0/sql-programming-guide.html

Cependant, dans de nombreux cas, les noms des paramètres peuvent être modifiés. Ceci pourrait causer des colonnes pour ne pas être repéré si le fichier n'a pas été mis à jour pour refléter le changement.

Comment puis-je spécifier une cartographie appropriée?

Je pense à quelque chose comme:

  val schema = StructType(Seq(
    StructField("name", StringType, nullable = false),
    StructField("age", IntegerType, nullable = false)
  ))


  val ps: Seq[Person] = ???

  val personRDD = sc.parallelize(ps)

  //Apply the schema to the RDD.
  val personDF: DataFrame = sqlContext.createDataFrame(personRDD, schema)
Malheureusement, il n'est pas clair ce que vous voulez. 1. Écrire avec parquet et arbitraire des noms? 2. Changer le parquet des noms de colonne par la suite? 3. Lire un parquet arbitraire des noms de colonne et de "correspondance"/carte jusqu'au champ de classe de cas?
Comment alors? Je veux mettre des noms de colonne manuellement et la carte de la classe de cas params dans ces colonnes.
Mais vous l'intention d'avoir leur correspondent automatiquement?
veuillez étendre sur ce point. Comme je l'ai dit je veux correspondre manuellement.

OriginalL'auteur BAR | 2015-09-12

  1. 8

    Fondamentalement, tout le mappage que vous devez faire peut être réalisé avec DataFrame.select(...). (Ici, je suppose, qu'aucun des conversions de type doivent être faites.)
    Compte tenu de l'avant et l'arrière de la cartographie, des cartes, l'essentiel est 

    val mapping = from.map{ (x:(String, String)) => personsDF(x._1).as(x._2) }.toArray
//personsDF your original dataframe  
val mappedDF = personsDF.select( mapping: _* )

    où la cartographie est un tableau de Columns avec alias.

    Exemple de code

    object Example {   
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkContext, SparkConf}
case class Person(name: String, age: Int)
object Mapping {
val from = Map("name" -> "a", "age" -> "b")
val to = Map("a" -> "name", "b" -> "age")
}
def main(args: Array[String]) : Unit = {
//init
val conf = new SparkConf()
.setAppName( "Example." )
.setMaster( "local[*]")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
import sqlContext.implicits._
//create persons
val persons = Seq(Person("bob", 35), Person("alice", 27))
val personsRDD = sc.parallelize(persons, 4)
val personsDF = personsRDD.toDF
writeParquet( personsDF, "persons.parquet", sc, sqlContext)
val otherPersonDF = readParquet( "persons.parquet", sc, sqlContext )
}
def writeParquet(personsDF: DataFrame, path:String, sc: SparkContext, sqlContext: SQLContext) : Unit = {
import Mapping.from
val mapping = from.map{ (x:(String, String)) => personsDF(x._1).as(x._2) }.toArray
val mappedDF = personsDF.select( mapping: _* )
mappedDF.write.parquet("/output/path.parquet") //parquet with columns "a" and "b"
}
def readParquet(path: String, sc: SparkContext, sqlContext: SQLContext) : Unit = {
import Mapping.to
val df = sqlContext.read.parquet(path) //this df has columns a and b
val mapping = to.map{ (x:(String, String)) => df(x._1).as(x._2) }.toArray
df.select( mapping: _* )
}
}

    Remarque

    Si vous avez besoin de convertir un dataframe retour à un EDR[Personne], puis

    val rdd : RDD[Row] = personsDF.rdd
val personsRDD : Rdd[Person] = rdd.map { r: Row => 
Person( r.getAs("person"), r.getAs("age") )
}

    Alternatives

    Ont également un oeil à Comment convertir étincelle SchemaRDD en CA de ma classe de cas?

    Belle approche. Pensez-vous que cela aurait un impact sur les performances, ou devrait-il pas être un facteur important puisque c'est compilé et optimisé une fois dans le pipeline interne?
    Je suppose que le dernier. Tout d'abord, comme il est Catalyseur d'optimisation / compilation. Deuxièmement, sélectionne (avec alias) ne semblent pas être des opérations coûteuses. Cependant, serait intéressé de voir certaines mesures de rendement ....

    OriginalL'auteur Martin Senne