Wie funktioniert HashPartitioner?

Lese ich bis auf die Dokumentation der HashPartitioner. Leider nichts wurde viel erklärt, außer für die API-Aufrufe. Ich bin unter der Annahme, dass HashPartitioner Partitionen verteilt, basierend auf dem Hashwert des keys. Zum Beispiel, wenn meine Daten wie

(1,1), (1,2), (1,3), (2,1), (2,2), (2,3)

So Partitionierer setzen würde, diese in verschiedene Partitionen mit gleichen Schlüsseln, die fallen in die gleiche partition. Aber ich verstehe nicht, die Bedeutung der Konstruktor-argument

new HashPartitoner(numPartitions) //What does numPartitions do?

Für die oben genannten dataset wie würden sich die Ergebnisse unterscheiden, wenn ich

new HashPartitoner(1)
new HashPartitoner(2)
new HashPartitoner(10)

Also, wie funktioniert HashPartitioner arbeiten die eigentlich?

InformationsquelleAutor der Frage Sohaib | 2015-07-15

107

Gut, wir machen das dataset unwesentlich mehr interessant:
```
val rdd = sc.parallelize(for {
    x <- 1 to 3
    y <- 1 to 2
} yield (x, None), 8)
```
Wir haben sechs Elemente:
```
rdd.count
```
```
Long = 6
```
kein Partitionierer:
```
rdd.partitioner
```
```
Option[org.apache.spark.Partitioner] = None
```
und acht Partitionen:
```
rdd.partitions.length
```
```
Int = 8
```
Können nun definieren, die kleinen Helfer zum zählen der Anzahl der Elemente pro partition:
```
import org.apache.spark.rdd.RDD

def countByPartition(rdd: RDD[(Int, None.type)]) = {
    rdd.mapPartitions(iter => Iterator(iter.length))
}
```
Da wir nicht Partitionierer unserem Datenbestand verteilt sich gleichmäßig zwischen den Partitionen (Standard-Partitionsschema Spark):
```
countByPartition(rdd).collect()
```
```
Array[Int] = Array(0, 1, 1, 1, 0, 1, 1, 1)
```
Jetzt können wir partitionieren unsere dataset:
```
import org.apache.spark.HashPartitioner
val rddOneP = rdd.partitionBy(new HashPartitioner(1))
```
Da parameter übergeben HashPartitioner definiert die Anzahl von Partitionen, die wir haben erwarten, dass Sie eine partition:
```
rddOneP.partitions.length
```
```
Int = 1
```
Da wir nur eine partition enthält alle Elemente:
```
countByPartition(rddOneP).collect
```
```
Array[Int] = Array(6)
```
Beachten Sie, dass die Reihenfolge der Werte nach dem mischen ist nicht-deterministisch.

Gleiche Weise, wenn wir HashPartitioner(2)
```
val rddTwoP = rdd.partitionBy(new HashPartitioner(2))
```
bekommen wir 2 Partitionen:
```
rddTwoP.partitions.length
```
```
Int = 2
```
Seit rdd ist durch key partitioniert Daten nicht gleichmäßig verteilt mehr:
```
countByPartition(rddTwoP).collect()
```
```
Array[Int] = Array(2, 4)
```
Weil mit drei Tasten und nur zwei verschiedene Werte von hashCode mod numPartitions es ist nichts unerwartetes hier:
```
(1 to 3).map((k: Int) => (k, k.hashCode, k.hashCode % 2))
```
```
scala.collection.immutable.IndexedSeq[(Int, Int, Int)] = Vector((1,1,1), (2,2,0), (3,3,1))
```
Nur um zu bestätigen, die oben:
```
rddTwoP.mapPartitions(iter => Iterator(iter.map(_._1).toSet)).collect()
```
```
Array[scala.collection.immutable.Set[Int]] = Array(Set(2), Set(1, 3))
```
Schließlich mit HashPartitioner(7) wir bekommen sieben Partitionen, drei nicht-leer mit 2 Elementen, je:
```
val rddSevenP = rdd.partitionBy(new HashPartitioner(7))
rddSevenP.partitions.length
```
```
Int = 7
```
```
countByPartition(rddTenP).collect()
```
```
Array[Int] = Array(0, 2, 2, 2, 0, 0, 0)
```
Zusammenfassung und Anmerkungen
- HashPartitioner übernimmt ein einzelnes argument, das definiert die Anzahl von Partitionen
- Werte zugewiesen werden Partitionen mit hash von Schlüsseln. hash Funktion kann je nach Sprache (Scala RDD verwenden kann hashCodeDataSets verwenden MurmurHash 3, PySpark, portable_hash).
  
  Im einfachen Fall wie diesem, wo der Schlüssel ist eine kleine Ganzzahl, können Sie davon ausgehen, dass hash ist eine Identität (i = hash(i)).
  
  Scala-API verwendet nonNegativeMod zu bestimmen partition basierend auf den berechneten hash,
- wenn die Verteilung der Tasten ist nicht einheitlich, Sie kann am Ende in Situationen, wenn ein Teil des Clusters ist im Leerlauf
- Tasten hashable. Sie können überprüfen, meine Antwort für Eine Liste als Schlüssel für PySpark ist reduceByKey zu Lesen über PySpark bestimmten Themen. Ein weiteres mögliches problem ist hervorgehoben durch HashPartitioner Dokumentation:
  
  Java-arrays haben hashCodes auf der Grundlage des arrays' Identitäten eher als Ihre Inhalte, also den Versuch eine partition RDD[Array[]] oder RDD[(Array[], _)] mit einem HashPartitioner produzieren eine unerwartete oder falsche Ergebnisse.
- In Python 3, die Sie haben, um sicherzustellen, dass das hashing ist in sich konsistent. Sehen Was bedeutet die Ausnahme: die Zufälligkeit der hash-Zeichenfolge sollte deaktiviert werden, über PYTHONHASHSEED bedeuten in pyspark?
- Hash-partitioner ist normalerweise weder Injektiv noch surjektive. Mehrere Schlüssel zugewiesen werden können, um eine einzelne partition und einige Partitionen leer bleiben.
- Bitte beachten Sie, dass derzeit hash-basierte Methoden gibt, die nicht in die Scala, wenn kombiniert mit REPL " angegebenen Fall Klassen (Fall die Gleichheit der Klassen in Apache Spark).
- HashPartitioner (oder jede andere Partitioner) mischt die Daten ein. Es sei denn, Partitionierung wiederverwendet wird zwischen mehreren Operationen es reduziert nicht die Menge an Daten, die gemischt werden.
InformationsquelleAutor der Antwort zero323
3

RDD vertrieben wird, das heißt, es ist aufgeteilt auf eine bestimmte Zahl der Teile. Jede dieser Partitionen wird möglicherweise auf andere Maschine. Hash-partitioner mit arument numPartitions choses, auf welcher partition platzieren-pair-Mädchen (key, value) im fallowing Weg:
1. Erzeugt genau numPartitions Partitionen.
2. Orte (key, value) in der partition mit der Nummer Hash(key) % numPartitions
InformationsquelleAutor der Antwort abalcerek
2

Den HashPartitioner.getPartition Methode nimmt einen Schlüssel als argument und gibt die index der partition, die der Schlüssel gehört. Der Partitionierer hat, um zu wissen, was die gültigen Indizes sind, so gibt es zahlen in der rechten Reihe. Die Anzahl der Partitionen, die angegeben ist durch die numPartitions Konstruktor-argument.

Die Implementierung gibt etwa key.hashCode() % numPartitions. Sehen Partitionierer.scala für mehr details.

InformationsquelleAutor der Antwort Daniel Darabos

Schreibe einen Kommentar

Du musst angemeldet sein, um einen Kommentar abzugeben.

Zusammenfassung und Anmerkungen