Comment puis-je créer une Étincelle DataFrame à partir d'un tableau imbriqué de struct element?

J'ai lu un fichier JSON dans Spark. Ce fichier a la structure suivante:

scala> tweetBlob.printSchema
root
 |-- related: struct (nullable = true)
 |    |-- next: struct (nullable = true)
 |    |    |-- href: string (nullable = true)
 |-- search: struct (nullable = true)
 |    |-- current: long (nullable = true)
 |    |-- results: long (nullable = true)
 |-- tweets: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- cde: struct (nullable = true)
...
...
 |    |    |-- cdeInternal: struct (nullable = true)
...
...
 |    |    |-- message: struct (nullable = true)
...
...

Ce que je voudrais idéalement veux, c'est un DataFrame avec des colonnes "cde", "cdeInternal", "message"... comme illustré ci-dessous

root
|-- cde: struct (nullable = true)
...
...
|-- cdeInternal: struct (nullable = true)
...
...
|-- message: struct (nullable = true)
...
...

J'ai réussi à utiliser une "explosion" d'en extraire des éléments les "tweets" tableau dans une colonne appelée "tweets"

scala> val tweets = tweetBlob.select(explode($"tweets").as("tweets"))
tweets: org.apache.spark.sql.DataFrame = [tweets: struct<cde:struct<author:struct<gender:string,location:struct<city:string,country:string,state:string>,maritalStatus:struct<evidence:string,isMarried:string>,parenthood:struct<evidence:string,isParent:string>>,content:struct<sentiment:struct<evidence:array<struct<polarity:string,sentimentTerm:string>>,polarity:string>>>,cdeInternal:struct<compliance:struct<isActive:boolean,userProtected:boolean>,tracks:array<struct<id:string>>>,message:struct<actor:struct<displayName:string,favoritesCount:bigint,followersCount:bigint,friendsCount:bigint,id:string,image:string,languages:array<string>,link:string,links:array<struct<href:string,rel:string>>,listedCount:bigint,location:struct<displayName:string,objectType:string>,objectType:string,postedTime...
scala> tweets.printSchema
root
|-- tweets: struct (nullable = true)
|    |-- cde: struct (nullable = true)
...
...
|    |-- cdeInternal: struct (nullable = true)
...
...
|    |-- message: struct (nullable = true)
...
...

Comment puis-je sélectionner toutes les colonnes à l'intérieur de la structure et de créer un DataFrame? Exploser ne fonctionne pas sur une struct si ma compréhension est correcte.

Toute aide est appréciée.

La Question a également demandé à Databricks forum, mais pas de réponse.

OriginalL'auteur zapstar | 2015-11-23

  1. 13

    Une façon possible de gérer cela est d'extraire les informations nécessaires à partir du schéma. Permet de commencer avec quelques données factices:

    import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.types._
case class Bar(x: Int, y: String)
case class Foo(bar: Bar)
val df = sc.parallelize(Seq(Foo(Bar(1, "first")), Foo(Bar(2, "second")))).toDF
df.printSchema
//root
// |-- bar: struct (nullable = true)
// |    |-- x: integer (nullable = false)
// |    |-- y: string (nullable = true)

    et une fonction d'assistance:

    def children(colname: String, df: DataFrame) = {
val parent = df.schema.fields.filter(_.name == colname).head
val fields = parent.dataType match {
case x: StructType => x.fields
case _ => Array.empty[StructField]
}
fields.map(x => col(s"$colname.${x.name}"))
}

    Enfin les résultats:

    df.select(children("bar", df): _*).printSchema
//root
//|-- x: integer (nullable = true)
//|-- y: string (nullable = true)
    Salut @zero323 il pourrait être une question stupide, mais est ce que les "enfants("bar", df): _*" cette syntaxe veux dire?
    children("bar", df) est juste un appel qui renvoie Seq[Column]. :_* effectue varargs déballage.
    merci pour l'explication
    Est-il un moyen de le faire dans SparkR?
    Il devrait être. SparkR schéma est équivalent à la Scala, et la syntaxe à point est universel pour tous les Spark SQL implémentations.

    OriginalL'auteur zero323

  2. 10

    Vous pouvez utiliser

    df
.select(explode(col("path_to_collection")).as("collection"))
.select(col("collection.*"))`:

    Exemple:

    scala> val json = """{"name":"Michael", "schools":[{"sname":"stanford", "year":2010}, {"sname":"berkeley", "year":2012}]}"""
scala> val inline = sqlContext.read.json(sc.parallelize(json :: Nil)).select(explode(col("schools")).as("collection")).select(col("collection.*"))
scala> inline.printSchema
root
|-- sname: string (nullable = true)
|-- year: long (nullable = true)
scala> inline.show
+--------+----+
|   sname|year|
+--------+----+
|stanford|2010|
|berkeley|2012|
+--------+----+

    Ou, vous pouvez également utiliser la fonction SQL inline:

    scala> val json = """{"name":"Michael", "schools":[{"sname":"stanford", "year":2010}, {"sname":"berkeley", "year":2012}]}"""
scala> sqlContext.read.json(sc.parallelize(json :: Nil)).registerTempTable("tmp")
scala> val inline = sqlContext.sql("SELECT inline(schools) FROM tmp")
scala> inline.printSchema
root
|-- sname: string (nullable = true)
|-- year: long (nullable = true)
scala> inline.show
+--------+----+
|   sname|year|
+--------+----+
|stanford|2010|
|berkeley|2012|
+--------+----+
    l'amélioration de votre réponse à m.w.r.t Spark 2.0. spark.lire.json(étincelle.createDataset(json :: Nil)).createOrReplaceTempView("tmp")
    Cette requête sql peut être utile à la solution ci-dessus """select nom,soit en ligne(écoles) de tmp"""

    OriginalL'auteur Fernando Lemos

  3. 4
    scala> import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.DataFrame
scala> import org.apache.spark.sql.types._
import org.apache.spark.sql.types._
scala> case class Bar(x: Int, y: String)
defined class Bar
scala> case class Foo(bar: Bar)
defined class Foo
scala> val df = sc.parallelize(Seq(Foo(Bar(1, "first")), Foo(Bar(2, "second")))).toDF
df: org.apache.spark.sql.DataFrame = [bar: struct<x: int, y: string>]
scala> df.printSchema
root
|-- bar: struct (nullable = true)
|    |-- x: integer (nullable = false)
|    |-- y: string (nullable = true)
scala> df.select("bar.*").printSchema
root
|-- x: integer (nullable = true)
|-- y: string (nullable = true)
scala>
    Cette méthode est beaucoup plus facile.

    OriginalL'auteur John