Wie funktioniert die Funktion Distinct () in Spark?

Ich bin ein Neuling auf Apache Spark und war lernen grundlegende Funktionalitäten.
Hatte einen kleinen Zweifel.Angenommen ich habe ein RDD von Tupeln (Schlüssel, Wert) - und wollte, um einige Einzelstücke aus Ihnen heraus. Benutze ich distinct() Funktion. Ich Frage mich, auf welcher Grundlage die Funktion betrachten, die Tupel als disparate..? Ist es auf der Grundlage der Schlüssel oder Werte oder beides?

InformationsquelleAutor der Frage preetham madeti | 2015-06-20

  1. 22

    .distinct() ist auf jeden Fall tun, einen shuffle in Partitionen. Um zu sehen, mehr von dem, was passiert ist, führen Sie einen .toDebugString auf Ihre RDD.

    val hashPart = new HashPartitioner(<number of partitions>)

val myRDDPreStep = <load some RDD>

val myRDD = myRDDPreStep.distinct.partitionBy(hashPart).setName("myRDD").persist(StorageLevel.MEMORY_AND_DISK_SER)
myRDD.checkpoint
println(myRDD.toDebugString)

    was für ein RDD Beispiel habe ich (myRDDPreStep schon ist hash-partitioniert nach Schlüssel beibehalten von StorageLevel.MEMORY_AND_DISK_SER und geprüfte), zurück:

    (2568) myRDD ShuffledRDD[11] at partitionBy at mycode.scala:223 [Disk Memory Serialized 1x Replicated]
+-(2568) MapPartitionsRDD[10] at distinct at mycode.scala:223 [Disk Memory Serialized 1x Replicated]
    |    ShuffledRDD[9] at distinct at mycode.scala:223 [Disk Memory Serialized 1x Replicated]
    +-(2568) MapPartitionsRDD[8] at distinct at mycode.scala:223 [Disk Memory Serialized 1x Replicated]
        |    myRDDPreStep ShuffledRDD[6] at partitionBy at mycode.scala:193 [Disk Memory Serialized 1x Replicated]
        |        CachedPartitions: 2568; MemorySize: 362.4 GB; TachyonSize: 0.0 B; DiskSize: 0.0 B
        |    myRDD[7] at count at mycode.scala:214 [Disk Memory Serialized 1x Replicated]

    Beachten Sie, dass es effizientere Möglichkeiten, um eine deutliche betreffen weniger mischt, vor ALLEM, wenn Ihr RDD ist bereits partitioniert in eine intelligente Art und Weise und die Partitionen sind nicht übermäßig verzerrt.

    Sehen Gibt es eine Möglichkeit zu umschreiben Funke RDD distinct zu verwenden, mapPartitions statt der klaren?
    und
    Apache Spark: Was ist das äquivalent Umsetzung von RDD.groupByKey() mit RDD.aggregateByKey()?

    InformationsquelleAutor der Antwort Glenn Strycker

  2. 8

    Den API docs für RDD.distinct() nur eine ein-Satz-Beschreibung:

    "Return eine neue RDD, mit der verschiedene Elemente in diesem RDD."

    Aus der jüngsten Erfahrung kann ich dir sagen, dass in einem Tupel-RDD das Tupel als ganzes betrachtet wird.

    Wenn Sie möchten, verschiedene Schlüssel oder unterschiedliche Werte, dann ist je nach genau das, was Sie erreichen möchten, können Sie entweder:

    A. call groupByKey() zu verwandeln {(k1,v11),(k1,v12),(k2,v21),(k2,v22)} zu {(k1,[v11,v12]), (k2,[v21,v22])} ; oder

    B. Streifen entweder der Schlüssel oder Werte durch aufrufen keys() oder values() gefolgt von distinct()

    Ab diesem schreiben (Juni 2015) UC Berkeley + EdX mit einer kostenlosen online-Kurs Einführung in Big Data und Apache Spark die Hände auf die Praxis mit diesen Funktionen.

    InformationsquelleAutor der Antwort Paul

  3. 6

    Justin Pihony richtig ist.Distinct verwendet die hashCode-und equals-Methode der Objekte für diese Bestimmung. Seine Rückkehr, die unterschiedliche Elemente(Objekt)

    val rdd = sc.parallelize(List((1,20), (1,21), (1,20), (2,20), (2,22), (2,20), (3,21), (3,22)))

    Verschiedene

    rdd.distinct.collect().foreach(println)
(2,22)
(1,20)
(3,22)
(2,20)
(1,21)
(3,21)

    Wenn Sie sich bewerben möchten verschiedene auf-Taste.
    In diesem Fall reduzieren, ist die bessere option

    ReduceBy

     val reduceRDD= rdd.map(tup =>
    (tup._1, tup)).reduceByKey { case (a, b) => a }.map(_._2)

reduceRDD.collect().foreach(println)

    Ausgabe:-

    (2,20)
(1,20)
(3,21)

    InformationsquelleAutor der Antwort Amit Dubey

  4. 4

    distinct verwendet die hashCode und equals - Methode der Objekte für diese Bestimmung. Tupel kommen, gebaut in und mit der Gleichheit Mechanismen delegieren von unten in die Gleichheit und die position jedes Objekts. Also distinct wird gegen die gesamte Tuple2 Objekt. Als Paul wies darauf hin, rufen Sie keys oder values und dann distinct. Oder schreiben Sie Ihre eigenen Werte über aggregateByKey, die würde halten Sie die Taste pairing. Oder, wenn Sie wollen, dass der eindeutige Schlüssel haben, dann könnten Sie eine regelmäßige aggregate

    InformationsquelleAutor der Antwort Justin Pihony

  5. 2

    Sieht es aus wie die distinct loswerden (Schlüssel, Wert) - Duplikate.

    In dem Beispiel weiter unten (1,20) und (2,20) zweimal wiederholt werden in myRDD, aber nach einer distinct(), werden die Duplikate entfernt.

    scala> val myRDD = sc.parallelize(List((1,20), (1,21), (1,20), (2,20), (2,22), (2,20), (3,21), (3,22)))
myRDD: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[1274] at parallelize at <console>:22

scala> myRDD.collect().foreach(println _)
(1,20)
(1,21)
(1,20)
(2,20)
(2,22)
(2,20)
(3,21)
(3,22)

scala> myRDD.distinct.collect().foreach(println _)
(2,22)
(1,20)
(3,22)
(2,20)
(1,21)
(3,21)

    InformationsquelleAutor der Antwort user3654449

Schreibe einen Kommentar