Spark DataFrame: ne groupBy après orderBy maintenir l'ordre?

J'ai une Spark 2.0 dataframe example avec la structure suivante:

id, hour, count
id1, 0, 12
id1, 1, 55
..
id1, 23, 44
id2, 0, 12
id2, 1, 89
..
id2, 23, 34
etc.

Il contient 24 entrées pour chaque id (une pour chaque heure de la journée) et est commandé par id, heure à l'aide de la orderBy fonction.

J'ai créé un Agrégateur groupConcat:

  def groupConcat(separator: String, columnToConcat: Int) = new Aggregator[Row, String, String] with Serializable {
    override def zero: String = ""

    override def reduce(b: String, a: Row) = b + separator + a.get(columnToConcat)

    override def merge(b1: String, b2: String) = b1 + b2

    override def finish(b: String) = b.substring(1)

    override def bufferEncoder: Encoder[String] = Encoders.STRING

    override def outputEncoder: Encoder[String] = Encoders.STRING
  }.toColumn

Il me permet de concaténer des colonnes dans les chaînes pour obtenir cette finale dataframe:

id, hourly_count
id1, 12:55:..:44
id2, 12:89:..:34
etc.

Ma question est, si je ne example.orderBy($"id",$"hour").groupBy("id").agg(groupConcat(":",2) as "hourly_count"), ne fait que garantir que l'horaire de compte sont classés correctement dans leurs seaux?

J'ai lu que ce n'est pas forcément le cas de Rdd (voir Étincelle de tri par clé, puis en groupe pour faire commandés itératif?), mais peut-être que c'est différent pour DataFrames ?

Si non, comment puis-je contourner cela ?

OriginalL'auteur Ana Todor | 2016-09-15

  1. 11

    groupBy après orderBy ne pas maintenir l'ordre, comme d'autres l'ont souligné. Ce que vous voulez faire est d'utiliser une fonction de Fenêtre--partition sur l'id et l'ordre par heures. Vous pouvez collect_list plus de cela, puis prendre le max (la plus grande) de la liste finale, car ils passent de façon cumulative (c'est à dire de la première heure ne se dans la liste, la deuxième heure aura 2 éléments dans la liste, et ainsi de suite).

    Exemple de code complet:

    import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.Window
import spark.implicits._

val data = Seq(( "id1", 0, 12),
  ("id1", 1, 55),
  ("id1", 23, 44),
  ("id2", 0, 12),
  ("id2", 1, 89),
  ("id2", 23, 34)).toDF("id", "hour", "count")

    val mergeList = udf{(strings: Seq[String]) => strings.mkString(":")}
    data.withColumn("collected", collect_list($"count")
                                                    .over(Window.partitionBy("id")
                                                                 .orderBy("hour")))
            .groupBy("id")
            .agg(max($"collected").as("collected"))
            .withColumn("hourly_count", mergeList($"collected"))
            .select("id", "hourly_count").show

    Ce qui nous garde dans le DataFrame monde. J'ai aussi simplifié l'UDF code de l'OP était de l'utiliser.

    De sortie:

    +---+------------+
| id|hourly_count|
+---+------------+
|id1|    12:55:44|
|id2|    12:89:34|
+---+------------+

    OriginalL'auteur Adair

  2. 5

    J'ai un cas où l'ordre n'est pas toujours tenue: parfois oui, surtout pas.

    Mon dataframe a 200 partitions en cours d'exécution sur l'Étincelle de 1,6

    df_group_sort = data.orderBy(times).groupBy(group_key).agg(
                                                  F.sort_array(F.collect_list(times)),
                                                  F.collect_list(times)
                                                           )

    pour vérifier la commande je compare les valeurs de retour de

    F.sort_array(F.collect_list(times))

    et

    F.collect_list(times)

    donnant par exemple (à gauche: sort_array(collect_list()); à droite: collect_list())

    2016-12-19 08:20:27.172000 2016-12-19 09:57:03.764000
2016-12-19 08:20:30.163000 2016-12-19 09:57:06.763000
2016-12-19 08:20:33.158000 2016-12-19 09:57:09.763000
2016-12-19 08:20:36.158000 2016-12-19 09:57:12.763000
2016-12-19 08:22:27.090000 2016-12-19 09:57:18.762000
2016-12-19 08:22:30.089000 2016-12-19 09:57:33.766000
2016-12-19 08:22:57.088000 2016-12-19 09:57:39.811000
2016-12-19 08:23:03.085000 2016-12-19 09:57:45.770000
2016-12-19 08:23:06.086000 2016-12-19 09:57:57.809000
2016-12-19 08:23:12.085000 2016-12-19 09:59:56.333000
2016-12-19 08:23:15.086000 2016-12-19 10:00:11.329000
2016-12-19 08:23:18.087000 2016-12-19 10:00:14.331000
2016-12-19 08:23:21.085000 2016-12-19 10:00:17.329000
2016-12-19 08:23:24.085000 2016-12-19 10:00:20.326000

    La colonne de gauche est toujours triée, tandis que la colonne de droite ne se compose que de tri de blocs.
    Pour les différentes exécutions de prendre(), l'ordre des blocs dans la colonne de droite est différente.

    La accepté de répondre a déclaré que vous avez besoin de trier par à la fois la colonne que vous souhaitez triés ainsi que des colonnes de votre groupe, c'est à dire orderBy(times, group_key).groupBy(group_key). Avez-vous essayé?

    OriginalL'auteur Kat

  3. 3

    Si vous voulez un travail autour de la mise en œuvre en Java (Scala et Python doit être similaire):

    example.orderBy(“hour”).groupBy(“id”).agg(functions.sort_array(functions.collect_list(functions.struct(dataRow.col(“hour”),dataRow.col(“count”))),false).as(“hourly_count”));

    OriginalL'auteur Shyam

  4. 1

    ordre peut ou peut ne pas être la même, en fonction du nombre de partitions et de la distribution des données. Nous pouvons résoudre en utilisant des rdd.

    Par exemple::

    J'ai sauvé la ci-dessous des exemples de données dans un fichier et l'a chargé dans hdfs.

    1,type1,300
2,type1,100
3,type2,400
4,type2,500
5,type1,400
6,type3,560
7,type2,200
8,type3,800

    et exécuté la commande ci-dessous:

    sc.textFile("/spark_test/test.txt").map(x=>x.split(",")).filter(x=>x.length==3).groupBy(_(1)).mapValues(x=>x.toList.sortBy(_(2)).map(_(0)).mkString("~")).collect()

    de sortie:

    Array[(String, String)] = Array((type3,6~8), (type1,2~1~5), (type2,7~3~4))

    Qui est, nous avons regroupé les données par type, par la suite, triées par prix, et la concaténation de l'ids par "~" comme séparateur.
    La commande ci-dessus peut être brisé comme ci-dessous:

    val validData=sc.textFile("/spark_test/test.txt").map(x=>x.split(",")).filter(x=>x.length==3)
val groupedData=validData.groupBy(_(1))  //group data rdds
val sortedJoinedData=groupedData.mapValues(x=>{
val list=x.toList
val sortedList=list.sortBy(_(2))
val idOnlyList=sortedList.map(_(0))
idOnlyList.mkString("~")
}
)
sortedJoinedData.collect()

    on peut alors prendre un groupe en particulier par l'aide de la commande

    sortedJoinedData.filter(_._1=="type1").collect()

    de sortie:

    Array[(String, String)] = Array((type1,2~1~5))

    OriginalL'auteur Ashish

  5. -4

    La réponse courte est Oui, le salaire horaire de compte, permettra de maintenir le même ordre.

    De généraliser, il est important que vous sorte avant de vous groupe. Aussi le tri doit être le même que le groupe + la colonne pour laquelle vous voulez vraiment le tri.

    Un exemple serait:

    employees
.sort("company_id", "department_id", "employee_role")
.groupBy("company_id", "department_id")
.agg(Aggregators.groupConcat(":", 2) as "count_per_role")
    Avez-vous des références indiquant que groupBy maintient l'ordre? Je ne pouvais pas trouver quelque chose dans les docs officielles
    Je n'ai pas de docs officielles, mais j'ai cette article qui explique un peu mieux le mécanisme de bzhangusc.wordpress.com/2015/05/28/... .Les commentaires sont intéressants aussi.
    Fait intéressant, même Sean Owen lui-même déclare que les commandes peuvent ne pas être conservés (issues.apache.org/jira/browse/...)
    Quelqu'un a lu l'article et les commentaires, j'ai ajouté le 7 juin 2017?

    OriginalL'auteur Interfector