Wie passend Dataframe Spaltennamen Scala Fall Attribute der Klasse?

Den Namen der Spalten in diesem Beispiel aus der spark-sql-kommen aus der 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

Jedoch in vielen Fällen die Namen der parameter geändert werden kann. Dies würde bewirken, dass die Spalten nicht gefunden werden, wenn die Datei wurde nicht aktualisiert, um die änderung widerzuspiegeln.

Wie kann ich festlegen, ein entsprechendes mapping?

Ich denke so etwas wie:

  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)
  • Es ist leider nicht klar, was Sie wollen. 1. Schreiben Parkett mit beliebigen Namen? 2. Ändern Sie die Parkett-Spaltennamen danach? 3. Lesen Sie ein Parkett mit beliebigen Spaltennamen und "match"/Karte zum Gebiet der case-Klasse?
  • Wie so? Ich möchte die set Spalte-Namen manuell und map case class params, um diese Spalten.
  • Aber Sie Absicht zu haben, Ihnen automatisch abgeglichen?
  • bitte erweitern Sie auf, dass. Wie ich sagte, ich will Spiel manuell.
InformationsquelleAutor BAR | 2015-09-12
  1. 8

    Grundsätzlich wird die Zuordnung, die Sie tun müssen, erreicht werden kann, mit DataFrame.select(...). (Ich nehme hier an, dass keine Konvertierungen durchgeführt werden müssen.)
    Angesichts der forward - und backward-mapping, maps, der wesentliche Teil ist 

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

    wo ist das mapping ein array von Columns mit alias.

    Beispiel-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: _* )
  }
}

    Bemerkung

    Wenn Sie brauchen, um zu konvertieren ein dataframe zurück zu einer RDD[Person], dann

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

    Alternativen

    Haben auch einen Blick auf Wie konvertieren von spark SchemaRDD in RDD meinem Fall Klasse?

    • Netter Ansatz. Glaubst du, dass dies Auswirkungen auf die Leistung, oder sollte es nicht ein Faktor sein, da diese kompiliert und optimiert wird einmal in der internen pipeline?
    • Ich nehme an, letzteres. Zuerst, als es ist-Katalysator-Optimierung / Zusammenstellung. Zweitens wählt (mit alias) scheint das nicht zu kostspielig. Obwohl, wäre daran interessiert zu sehen, einige performance-Messungen ....
    • können wir jave hier ? für das Beispiel gegeben? In java-dataset select-Methode haben nicht die Fähigkeit, eine map?
    InformationsquelleAutor Martin Senne
Schreibe einen Kommentar