Dans Apache Spark, pourquoi ne RDD.de l'union de ne pas préserver l'outil de partitionnement?

Comme tout le monde le sait partitioners Spark ont un énorme impact sur les performances sur un "large" des opérations, de sorte qu'il est généralement personnalisé dans les opérations. J'ai essayé avec le code suivant:

val rdd1 =
  sc.parallelize(1 to 50).keyBy(_ % 10)
    .partitionBy(new HashPartitioner(10))
val rdd2 =
  sc.parallelize(200 to 230).keyBy(_ % 13)

val cogrouped = rdd1.cogroup(rdd2)
println("cogrouped: " + cogrouped.partitioner)

val unioned = rdd1.union(rdd2)
println("union: " + unioned.partitioner)

Je vois que, par défaut, cogroup() donne toujours un EDR avec l'outil de partitionnement personnalisé, mais union() ne le fait pas, il sera toujours en revenir à la valeur par défaut. C'est contre-intuitif que nous avons l'habitude de supposer qu'un PairRDD devrait user de son premier élément clé de partition. Est-il un moyen de "forcer" l'Étincelle de fusionner 2 PairRDDs à utiliser la même clé de partition?

InformationsquelleAutor tribbloid | 2015-04-30
  1. 40

    union est très efficace, car il ne bouge pas toutes les données. Si rdd1 a 10 partitions et rdd2 a 20 partitions puis rdd1.union(rdd2) aura 30 partitions: les partitions des deux RDDs mis les uns après les autres. C'est juste une tenue de la comptabilité de changement, il n'y a pas de shuffle.

    Mais forcément, il ignore l'outil de partitionnement. Un outil de partitionnement est construit pour un nombre donné de partitions. Le résultant CA a un certain nombre de partitions qui est différent des deux rdd1 et rdd2.

    Après la prise de l'union, vous pouvez exécuter repartition aléatoire les données, organisation des données par clé.

    Il existe une exception à ce qui précède. Si rdd1 et rdd2 ont le même outil de partitionnement (avec le même nombre de partitions), union se comporte différemment. Il rejoindra les partitions des deux Rdd par paires, en lui donnant le même nombre de partitions que chacune des entrées avaient. Cela peut nécessiter de déplacer les données (si les partitions ne sont pas co-situé), mais n'impliquera pas la lecture aléatoire. Dans ce cas, l'outil de partitionnement est conservé. (Le code est dans PartitionerAwareUnionRDD.scala.)

    • Il y a en fait un outil de partionnement-conscient de l'union RDD qui, je pense, est censé être utilisé automatiquement dans les cas où le partitionnement peut être conservé; pas sûr pourquoi elle n'est pas appliquée ici. Voir github.com/apache/spark/blob/... et github.com/apache/spark/blob/master/core/src/main/scala/org/...
    • Wow, cool! Ne savait jamais sur qui. On dirait qu'il est utilisé uniquement lorsque les deux Rdd ont le même outil de partitionnement. Je vais l'ajouter à la réponse, merci!
    • Merci beaucoup! C'est un élément très important de l'optimisation. BTW, si ce n'est pas optimal pour tous les cas, je peux toujours écrire un zip + dans la partition de l'union de toute façon
    • Excellente réponse de Daniel. Je vous remercie.
    • Très intéressant!!! Est-il une manière spécifique de s'assurer qu'ils ont le même programme de partitionnement et le même nombre de partitions(sans repartitionner)? Je suis l'exécution itérative syndicats de Dataframes (bigDF.de l'union(oneRowDF) de manière itérative) avec pyspark.
    • Presque tout ce qui utilise un HashPartitioner. Donc, si votre DataFrames ont le même nombre de partitions, j'espère que ce serait suffisant. Vous pouvez simplement imprimer df.partitioner et df.partitions pour voir ce qui se passe.
    • Juste pour ajouter que les commandes correctes sont df.rdd.partitioner et df.rdd.getNumPartitions. N'ai aucune idée de pourquoi mon DFs n'ont pas d'outil de partitionnement (Aucun) même quand je suis repartitionnement eux?
    • Ah, désolé, j'étais tout à fait tort. Cela ne s'applique pas à DataFrames à tous. Vous avez besoin d'une clé pour le partitionnement. DataFrames n'avez pas les clés. Aussi union peut être complètement différente pour eux que pour les Rdd. Désolé pour vous induire en erreur.
    • Si votre oneRowDF n'a en fait qu'une ligne, peut-être que vous pourriez essayer de recueillir tous les de local et de la construction d'une taille raisonnable DF d'eux avant d'aller de l'union. (Je n'ai pas essayé.)

    InformationsquelleAutor Daniel Darabos
  2. 2

    Ce n'est plus vrai. Iff deux Rdd ont exactement le même programme de partitionnement et le nombre de partitions, le unioned CA va aussi avoir les mêmes partitions. Cela a été introduit dans https://github.com/apache/spark/pull/4629 et incorporé dans Spark 1.3.

    InformationsquelleAutor Joel Croteau