Spark dataframe exploser fonction

Quelqu'un peut-il expliquer pourquoi les cas Row, Seq[Row] sont utilisés après l'exploser d'un dataframe champ de la collecte des éléments.
Et aussi pouvez-vous svp m'expliquer la raison pour laquelle asInstanceOf est nécessaire pour obtenir les valeurs de a explosé champ?

Voici la syntaxe:

val explodedDepartmentWithEmployeesDF = departmentWithEmployeesDF.explode(departmentWithEmployeesDF("employees")) {     
                          case Row(employee: Seq[Row]) => 
                          employee.map(employee =>
                          Employee(employee(0).asInstanceOf[String], 
                          employee(1).asInstanceOf[String], employee(2).asInstanceOf[String]) ) }
InformationsquelleAutor Ramesh | 2016-08-24
  1. 2

    D'abord je note, que je ne peux pas expliquer pourquoi votre explode() se transforme en Row(employee: Seq[Row]) que je ne connais pas le schéma de votre DataFrame. Je suppose qu'il a à faire avec la structure de vos données.

    Ne vous connaissant pas les données d'origine, j'ai créé un petit jeu de données pour des travaux de

    scala> val df = sc.parallelize( Array( (1, "dsfds dsf dasf dsf dsf d"), (2, "2344 2353 24 23432 234"))).toDF("id", "text")
df: org.apache.spark.sql.DataFrame = [id: int, text: string]

    Si j'ai maintenant une carte sur elle, vous pouvez voir qu'il renvoie les lignes contenant des données de type any.

    scala> df.map {case row: Row => (row(0), row(1)) }
res21: org.apache.spark.rdd.RDD[(Any, Any)] = MapPartitionsRDD[17] at map at <console>:33

    Vous avez perdu les informations de type, qui est pourquoi vous devez spécifier explicitement le type lorsque vous souhaitez utiliser les données de la ligne

    scala> df.map {case row: Row => (row(0).asInstanceOf[Int], row(1).asInstanceOf[String]) }
res22: org.apache.spark.rdd.RDD[(Int, String)] = MapPartitionsRDD[18] at map at <console>:33

    Donc, pour exploser, je dois faire la suite

    scala> :paste
//Entering paste mode (ctrl-D to finish)

import org.apache.spark.sql.Row
df.explode(col("id"), col("text")) {case row: Row =>
    val id = row(0).asInstanceOf[Int]
    val words = row(1).asInstanceOf[String].split(" ")
    words.map(word => (id, word))
}

//Exiting paste mode, now interpreting.

import org.apache.spark.sql.Row
res30: org.apache.spark.sql.DataFrame = [id: int, text: string, _1: int, _2: string]

scala> res30 show
+---+--------------------+---+-----+
| id|                text| _1|   _2|
+---+--------------------+---+-----+
|  1|dsfds dsf dasf ds...|  1|dsfds|
|  1|dsfds dsf dasf ds...|  1|  dsf|
|  1|dsfds dsf dasf ds...|  1| dasf|
|  1|dsfds dsf dasf ds...|  1|  dsf|
|  1|dsfds dsf dasf ds...|  1|  dsf|
|  1|dsfds dsf dasf ds...|  1|    d|
|  2|2344 2353 24 2343...|  2| 2344|
|  2|2344 2353 24 2343...|  2| 2353|
|  2|2344 2353 24 2343...|  2|   24|
|  2|2344 2353 24 2343...|  2|23432|
|  2|2344 2353 24 2343...|  2|  234|
+---+--------------------+---+-----+

    Si vous voulez nommé colonnes, vous pouvez définir une classe de cas de vous tenir explosé données

    scala> :paste
//Entering paste mode (ctrl-D to finish)

import org.apache.spark.sql.Row
case class ExplodedData(word: String)
df.explode(col("id"), col("text")) {case row: Row =>
    val words = row(1).asInstanceOf[String].split(" ")
    words.map(word => ExplodedData(word))
}

//Exiting paste mode, now interpreting.

import org.apache.spark.sql.Row
defined class ExplodedData
res35: org.apache.spark.sql.DataFrame = [id: int, text: string, word: string]

scala> res35.select("id","word").show
+---+-----+
| id| word|
+---+-----+
|  1|dsfds|
|  1|  dsf|
|  1| dasf|
|  1|  dsf|
|  1|  dsf|
|  1|    d|
|  2| 2344|
|  2| 2353|
|  2|   24|
|  2|23432|
|  2|  234|
+---+-----+

    Espère que cela apporte une certaine clearity.

    InformationsquelleAutor Andreas Ryge
  2. 0

    Je pense que vous pouvez en savoir plus sur le document et faites un test au préalable.

    exploser d'un dataframe encore de retour d'un dataframe.Et il accepte une fonction lambda f: (Ligne) ⇒ TraversableOnce[A] comme paramètre.

    dans la fonction lambda, vous ferez correspondre l'entrée par cas. Vous avez déjà connu que votre contribution sera la Ligne de l'employé, qui est encore un Suivants de la Ligne.De sorte que le cas de l'entrée Ligne(employé: Seq[Ligne]) , si vous ne comprenez pas cette partie, vous pouvez apprendre plus de chose sur délettrer d'une fonction en scala.

    Et que, employé(je crois que vous devriez utiliser employés ici), en tant que Seq de Ligne, s'appliquent à la carte en fonction de la carte chaque ligne à un Employé. Et vous pourrez utiliser la scala appliquer la fonction pour obtenir le i-ième valeur de cette ligne. Mais la valeur de retour est un Objet , vous devez utiliser des asInstanceOf pour le convertir dans le type que vous avez prévu.

    InformationsquelleAutor John Zeng