Schiefe dataset mitmachen Funke?

Ich bin das verbinden von zwei großen Datensätzen mithilfe von Spark RDD. Ein dataset ist sehr viel schief, so dass einige der Testamentsvollstrecker Aufgaben nehmen eine lange Zeit, um den job zu beenden. Wie kann ich lösen dieses Szenario?

InformationsquelleAutor Raj Kumar | 2016-11-02
  1. 19

    Ziemlich guten Artikel über, wie es getan werden kann: https://datarus.wordpress.com/2015/05/04/fighting-the-skew-in-spark/

    Kurze version:

    • Hinzufügen von zufälligen element, um große RDD und neue join-Schlüssel (mit dem es
    • Hinzufügen von zufälligen element, um kleine RDD mit explodieren/flatMap Erhöhung der Zahl der Einträge und neue join-Schlüssel
    • Join RDDs auf neue join-Schlüssel, die jetzt verteilt werden besser durch random seeding
    • Ging durch die Artikel, die Sie erwähnt. In meinem Fall, größeren Tisch hat 23500 Partitionen mit 2 Milliarden Datensätze. Kleinere Tisch hat 5 Millionen Datensätze. Wie ich entscheiden, den Wert 'N'? Irgendwelche Vorschläge?
    • es ist eine weitere parameter, die berücksichtigt werden sollen, die Größe des shuffle-block in der Funke sollte weniger als 2 GB SPARK-6235. Ich würde empfehlen, den Fokus auf die Größe der partition, die normalerweise sein sollte ~128MB statt der Anzahl der Partitionen. Ich habe gesehen-Anwendungen arbeiten gut mit vielen Partitionen (> 32 Kb).
    • Sowie Sie möchten, um genügend Partitionen auf der neuen Schlüssel, die alle Ihre Vollzieher erhalten Daten, um beitreten zu maximieren Parallelität.
    • es ist etwas, was ich nicht bekommen, in der python-code small_rdd_transformed = small_rdd.cartesian(sc.parallelize(range(0, N))).map(lambda x: ((x[0][0], x[1]), x[0][1])).coalesce(num_parts).cache() # replicate the small rdd im scala-x ist ein tuple2, was so tut, x[0][0] bedeutet beispielsweise?..
    • mein Fehler, es ist ein mapPartitionWithIndex fehlt
    InformationsquelleAutor LiMuBei
  2. 10

    Je nach der bestimmten Art neigen Sie erleben, möglicherweise gibt es verschiedene Möglichkeiten, es zu lösen. Die Grundidee ist:

    • Ändern Sie die join-Spalte, oder erstellen Sie einen neuen join-Spalte, die ist nicht schief, sondern die noch immer ausreichende Informationen für die join -
    • Tun die Verknüpfung auf, dass nicht-schiefe Spalte -- resultierenden Partitionen werden nicht verzerrt
    • Folgenden die beitreten möchten, können Sie die join-Spalte zurück, um Ihre bevorzugte format, oder legen Sie, wenn Sie erstellt eine neue Spalte

    Den "Kampf der Skew In der Funke" Artikel verwiesen, die in LiMuBei s Antwort ist eine gute Technik, wenn die schiefe der Daten nimmt in der Verknüpfung. In meinem Fall, skew verursacht wurde durch eine sehr große Anzahl von null-Werten im join-Spalte. Die null-Werte wurden nicht an die anzuschließen, aber da Spark Partitionen auf die join-Spalte, die post-join Partitionen waren sehr schief sein, wie es war eine riesige partition, die alle von der null-Werte.

    Ich löste es durch hinzufügen einer neuen Spalte, die geändert, werden alle null-Werte zu einem gut verteilten temporären Wert, wie "NULL_VALUE_X", wobei X ersetzt wird durch zufällige zahlen zwischen 1 und 10.000, z.B. (in Java):

    //Before the join, create a join column with well-distributed temporary values for null swids.  This column
//will be dropped after the join.  We need to do this so the post-join partitions will be well-distributed,
//and not have a giant partition with all null swids.
String swidWithDistributedNulls = "swid_with_distributed_nulls";
int numNullValues = 10000; //Just use a number that will always be bigger than number of partitions
Column swidWithDistributedNullsCol =
    when(csDataset.col(CS_COL_SWID).isNull(), functions.concat(
        functions.lit("NULL_SWID_"),
        functions.round(functions.rand().multiply(numNullValues)))
    )
    .otherwise(csDataset.col(CS_COL_SWID));
csDataset = csDataset.withColumn(swidWithDistributedNulls, swidWithDistributedNullsCol);

    Dann der Beitritt auf diese neue Spalte und dann nach der Verknüpfung:

    outputDataset.drop(swidWithDistributedNullsCol);
    InformationsquelleAutor Jason Evans
  3. 10

    Sagen, Sie haben zum verknüpfen von zwei Tabellen A und B on A. id=B. id. Nehmen wir an, Eine Tabelle hat die skew auf id=1.

    d.h. select A. id from A join B on A. id = B. id

    Gibt es zwei grundlegende Ansätze zur Lösung der skew join Problem:

    Ansatz 1:

    Break your query/dataset in 2 Teile - mit nur neigen und die anderen, die nicht verzerrte Daten.
    In dem oben genannten Beispiel. Abfrage wird -

     1. select A.id from A join B on A.id = B.id where A.id <> 1;
 2. select A.id from A join B on A.id = B.id where A.id = 1 and B.id = 1;

    Die erste Abfrage wird sich nicht neigen, so dass alle Aufgaben von ResultStage fertig zu ungefähr der gleichen Zeit.

    Wenn wir davon ausgehen, dass B nur einige wenige Zeilen mit B. id = 1, dann wird es passen in den Speicher. So den Zweiten Abfrage werden in einer Sendung mitmachen. Dies wird auch als Map-side-join-in-Struktur.

    Referenz: https://cwiki.apache.org/confluence/display/Hive/Skewed+Join+Optimierung

    Die teilweise die Ergebnisse der beiden Abfragen können dann zusammengeführt werden, um die endgültigen Ergebnisse.

    Ansatz 2:

    Auch erwähnt LeMuBei oben, der 2. Ansatz versucht, mischen Sie die join-Schlüssel durch anfügen von zusätzlichen Spalte.
    Schritte:

    1. Fügen Sie eine Spalte in der größeren Tabelle (A), sagen skewLeft und füllen es mit Zufallszahlen zwischen 0 bis N-1 für alle Zeilen.

    2. Fügen Sie eine Spalte in der kleineren Tabelle (B), sagen skewRight. Replizieren Sie den kleineren Tisch N-mal. Also Werte in neuen skewRight Spalte variiert von 0 bis N-1 für jede Kopie der original-Daten. Für diese, die Sie verwenden können, die explodieren sql/dataset Betreiber.

    Nach 1 und 2, kommen die 2 Datensätzen/Tabellen mit join-Bedingung aktualisiert-

                    *A.id = B.id && A.skewLeft = B.skewRight*

    Referenz: https://datarus.wordpress.com/2015/05/04/fighting-the-skew-in-spark/

    • Wäre toll, wenn Sie zur Erarbeitung der zweite Ansatz mit einigen einfachen Beispiel dataframes. Ich brauche, um dies zu implementieren, in Pyspark. Dank
    InformationsquelleAutor prakharjain
  4. -1

    Könnten Sie versuchen, neu zu partitionieren die "schiefe" RDD mehr Partitionen, oder erhöhen Sie die spark.sql.shuffle.partitions (das ist standardmäßig 200).

    In deinem Fall würde ich versuchen, die Anzahl der Partitionen zu werden, viel höher als die Zahl der Vollzieher.

    • spark.sql.shuffle.Partitionen werden nicht helfen, verzerrt die Daten. Es werden 200-Partitionen, aber nur wenige von Ihnen haben die Daten.
    • Erhöhung der Funke.sql.shuffle.Partitionen zu einer höheren Anzahl nicht helfen mit Skew. Alle Daten, die entsprechend schiefe Taste gehen noch zur gleichen reducer und Ursache für die Langsamkeit.
    • das ist nicht ganz richtig. Erhöhung der Anzahl von Partitionen wird die Wahrscheinlichkeit reduziert, dass zwei Schlüssel mit vielen Datensatz wird in der gleichen partition.
    • spark.sql.shuffle.Partitionen wird nicht den trick tun. mitmachen bei schräg liegenden Daten führen hot spotting Problem auf Daten mischen, weil der gleiche Wert auf der join point wird gehashed in der gleichen hash-Schlüssel und aufgrund der, dass alle diese Zeilen(mit den gleichen Wert) wird an den gleichen executor während des Mischens. siehe: oreilly.com/library/view/high-performance-spark/9781491943199/...
    InformationsquelleAutor Raphael Roth
Schreibe einen Kommentar