Quelle est la différence entre les rejoindre et cogroup dans Apache Spark

Quelle est la différence entre les rejoindre et cogroup dans Apache Spark? Ce qui est le cas d'utilisation de chaque méthode?

InformationsquelleAutor miaoiao | 2017-05-14
  1. 48

    Laissez-moi vous aider à les clarifier, les deux sont communs pour l'utilisation et la important de!

    def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]

    C'est prototype de rejoindre, veuillez regarder attentivement il. Par exemple,

    val rdd1 = sc.makeRDD(Array(("A","1"),("B","2"),("C","3")),2)
val rdd2 = sc.makeRDD(Array(("A","a"),("C","c"),("D","d")),2)

scala> rdd1.join(rdd2).collect
res0: Array[(String, (String, String))] = Array((A,(1,a)), (C,(3,c)))

    Toutes les touches qui apparaissent dans le résultat final est commun à rdd1 et rdd2. Ceci est similaire à relation database operation INNER JOIN.

    Mais cogroup est différent,

    def cogroup[W](other: RDD[(K, W)]): RDD[(K, (Iterable[V], Iterable[W]))]

    est l'une des clés, au moins apparaître dans l'un des deux rdd, il apparaît dans le résultat final, permettez-moi de préciser:

    val rdd1 = sc.makeRDD(Array(("A","1"),("B","2"),("C","3")),2)
val rdd2 = sc.makeRDD(Array(("A","a"),("C","c"),("D","d")),2)

scala> var rdd3 = rdd1.cogroup(rdd2).collect
res0: Array[(String, (Iterable[String], Iterable[String]))] = Array(
(B,(CompactBuffer(2),CompactBuffer())), 
(D,(CompactBuffer(),CompactBuffer(d))), 
(A,(CompactBuffer(1),CompactBuffer(a))), 
(C,(CompactBuffer(3),CompactBuffer(c)))
)

    C'est très similar à relation database operation FULL OUTER JOIN, mais au lieu d'aplatir le résultat par ligne par enregistrement, il vous donnera la interable interface vous, l'opération suivante est à vous en tant que pratique!

    Bonne Chance!

    Étincelle docs est: http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.rdd.PairRDDFunctions

    • Comment utiliser cogroup pour grands ensembles de données, comme lorsque j'utilise collect() il jette de la mémoire exception rdd1 = rdd2.cogroup(rdd3).collect. Pouvez-vous aider sur cette question [stackoverflow.com/questions/47180307/... vous
    • veuillez utiliser la méthode de prendre au lieu de rassembler.
    • yh, essayé avec take, top savez-vous comment repartitionner le code ci-dessous
    • val source_primary_key = s.map(rec => (rec.split(",")(0), rec)) val destination_primary_key = k.map(rec => (rec.split(",")(0), rec)) val cos = source_primary_key.cogroup(destination_primary_key)
    • def cogroup[W](autres: RDD[(K, W)], numPartitions: Int): CA[(K,(Itératif[V], Itératif[W]))], vous pouvez directement ajouter un second paramètre de la cogroup méthode.
    • Est "cogroup" moins cher que la jointure externe complète? Semble comme cogroup juste recueille des entrées avec la même clé à la même tâche. Je voudrais obtenir la pleine shuffle lire, dans le pire des cas, non?

    InformationsquelleAutor ashburshui