L'agrégation de plusieurs colonnes avec une fonction personnalisée dans l'étincelle

Je me demandais si il est possible de spécifier une coutume fonction d'agrégation pour le spark dataframes sur plusieurs colonnes.

J'ai un tableau comme celui du type (nom, article, prix):

john | tomato | 1.99
john | carrot | 0.45
bill | apple  | 0.99
john | banana | 1.29
bill | taco   | 2.59

:

Je voudrais globale de l'élément et c'est un coût pour chaque personne dans une liste comme ceci:

john | (tomato, 1.99), (carrot, 0.45), (banana, 1.29)
bill | (apple, 0.99), (taco, 2.59)

Est-ce possible dans dataframes? J'ai récemment appris à propos de collect_list mais il semble ne fonctionne que pour une seule colonne.

OriginalL'auteur anthonybell | 2016-06-09

  1. 25

    La façon la plus simple de le faire comme un DataFrame est d'abord de recueillir des deux listes, et ensuite utiliser un UDF à zip les deux listes ensemble. Quelque chose comme:

    import org.apache.spark.sql.functions.{collect_list, udf}
import sqlContext.implicits._

val zipper = udf[Seq[(String, Double)], Seq[String], Seq[Double]](_.zip(_))

val df = Seq(
  ("john", "tomato", 1.99),
  ("john", "carrot", 0.45),
  ("bill", "apple", 0.99),
  ("john", "banana", 1.29),
  ("bill", "taco", 2.59)
).toDF("name", "food", "price")

val df2 = df.groupBy("name").agg(
  collect_list(col("food")) as "food",
  collect_list(col("price")) as "price" 
).withColumn("food", zipper(col("food"), col("price"))).drop("price")

df2.show(false)
# +----+---------------------------------------------+
# |name|food                                         |
# +----+---------------------------------------------+
# |john|[[tomato,1.99], [carrot,0.45], [banana,1.29]]|
# |bill|[[apple,0.99], [taco,2.59]]                  |
# +----+---------------------------------------------+
    Bonne réponse ! 🙂
    J'ai utilisé col(...) au lieu de $"..." pour une bonne raison, je trouve col(...) fonctionne avec moins de travail à l'intérieur des choses comme class définitions.
    La réponse suppose (peut-être correctement) que collect_list() permet de conserver l'ordre des éléments sur les deux colonnes de la nourriture & prix. Ce qui signifie que la nourriture et des prix à partir de la même ligne prendra fin à la même indice dans les deux recueillies listes. Est-ce afin de préserver le comportement de la garantie? (cela aurait du sens, mais je ne suis pas sûr qu'en regardant la scala de code pour collect_list, pas un scala programmeur).
    Autant que je sache, il n'y a aucune garantie que l'ordre des éléments sera le même. cf : stackoverflow.com/questions/40407514/...
    J'ai utilisé une variante de cette solution à zip cinq listes ensemble. Cela m'a donné l'occasion d'écrire la meilleure ligne de code de ma carrière jusqu'à présent: _ zip _ zip _ zip _ zip _

    OriginalL'auteur David Griffin

  2. 39

    Envisager l'utilisation de la struct fonction de groupe de l'ensemble des colonnes avant de recueillir une liste:

    import org.apache.spark.sql.functions.{collect_list, struct}
import sqlContext.implicits._

val df = Seq(
  ("john", "tomato", 1.99),
  ("john", "carrot", 0.45),
  ("bill", "apple", 0.99),
  ("john", "banana", 1.29),
  ("bill", "taco", 2.59)
).toDF("name", "food", "price")

df.groupBy($"name")
  .agg(collect_list(struct($"food", $"price")).as("foods"))
  .show(false)

    Sorties:

    +----+---------------------------------------------+
|name|foods                                        |
+----+---------------------------------------------+
|john|[[tomato,1.99], [carrot,0.45], [banana,1.29]]|
|bill|[[apple,0.99], [taco,2.59]]                  |
+----+---------------------------------------------+
    Je tiens à préciser que cette approche semble plus propre que l'on a accepté la réponse, mais malheureusement ne fonctionne pas avec spark 1.6, parce que collect_list() ne pas accepter un struct.
    Œuvres Spark 2.1

    OriginalL'auteur Daniel Siegmann

  3. 2

    Ici est une option par la conversion de la trame de données pour un CA de Carte et ensuite appeler une groupByKey sur elle. Le résultat sera une liste de paires clé-valeur où la valeur est une liste de tuples.

    df.show
+----+------+----+
|  _1|    _2|  _3|
+----+------+----+
|john|tomato|1.99|
|john|carrot|0.45|
|bill| apple|0.99|
|john|banana|1.29|
|bill|  taco|2.59|
+----+------+----+


val tuples = df.map(row => row(0) -> (row(1), row(2)))
tuples: org.apache.spark.rdd.RDD[(Any, (Any, Any))] = MapPartitionsRDD[102] at map at <console>:43

tuples.groupByKey().map{ case(x, y) => (x, y.toList) }.collect
res76: Array[(Any, List[(Any, Any)])] = Array((bill,List((apple,0.99), (taco,2.59))), (john,List((tomato,1.99), (carrot,0.45), (banana,1.29))))

    OriginalL'auteur Psidom

  4. 2

    Peut-être une meilleure façon que les zip fonction (depuis l'UDF et l'udaf, nous sont très mauvais pour la performance) est à envelopper les deux colonnes en Struct.

    Ce serait probablement travailler aussi bien:

    df.select('name, struct('food, 'price).as("tuple"))
  .groupBy('name)
  .agg(collect_list('tuple).as("tuples"))

    OriginalL'auteur Yifan Guo