Wenn Sie mit großen Datensätzen in Ein gemeinsames Leistungsproblem ist Dies geschieht, wenn ein paar Schlüssel Die Datenverteilung führt zu Partitionen und langsame Abfragen. Dies geschieht hauptsächlich während von Operationen, die So wie Oder sogar regelmäßig . Apache Spark data skew dominate uneven shuffling joins aggregations Ein praktischer Weg, um den Schwanz zu reduzieren, ist , was die künstliche Verbreitung schwerer Schlüssel über mehrere Partitionen beinhaltet. In diesem Beitrag werde ich Sie durch dies mit einem praktischen Beispiel führen. salting Wie Salting Data Skew-Probleme löst Durch Hinzufügen a Wir können große Schlüssel gleichmäßiger verteilen, indem wir die erzeugte Zahl auf den Join-Schlüssel übertragen und dann über diesen kombinierten Schlüssel übertragen.Dies macht die Datenverteilung einheitlicher und verteilt die Last auf mehr Arbeiter, anstatt den Großteil der Daten an einen Arbeiter zu senden und die anderen leer zu lassen. randomly Vorteile des Salzens 
 
 
 
 
 
  Spreads data evenly across partitions, preventing a few workers overload and improves utilization. Reduced Skew: 
 
  Speeds up joins and aggregations by balancing the workload. Improved Performance: 
 
  Reduces the risk of out-of-memory errors caused by large, uneven partitions. Avoids Resource Contention: Wann Salz verwenden Bei Joins oder Aggregationen mit verzerrten Tasten verwenden Sie Salting, wenn Sie lange Shuffle-Zeiten oder Ausführungsfehler aufgrund von Datenverschiebungen bemerken. Es ist auch hilfreich in Echtzeit-Streaming-Anwendungen, bei denen die Partitionierung die Effizienz der Datenverarbeitung beeinflusst, oder wenn die meisten Mitarbeiter leer sind, während einige in einem laufenden Zustand stecken. Ein Beispiel in der Scala Lassen Sie uns einige Daten mit einem Wir können annehmen, dass es zwei Datensätze gibt, die wir verbinden müssen: Eine ist ein großer Datensatz und die andere ist ein kleiner Datensatz. unbalanced import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

// Simulated large dataset with skew
val largeDF = Seq(
  (1, "txn1"), (1, "txn2"), (1, "txn3"), (2, "txn4"), (3, "txn5")
).toDF("customer_id", "transaction")

// Small dataset
val smallDF = Seq(
  (1, "Ahmed"), (2, "Ali"), (3, "Hassan")
).toDF("customer_id", "name")
 Fügen wir die Salzspalte zu den großen Datensätzen hinzu, die wir verwenden um die Werte des großen Schlüssels in kleinere Partitionen zu verteilen randomization // Step 1: create a salting key in the large dataset
val numBuckets = 3
val saltedLargeDF = largeDF.
    withColumn("salt", (rand() * numBuckets).cast("int")).
    withColumn("salted_customer_id", concat($"customer_id", lit("_"), $"salt"))

saltedLargeDF.show()
+-----------+-----------+----+------------------+
|customer_id|transaction|salt|salted_customer_id|
+-----------+-----------+----+------------------+
|          1|       txn1|   1|               1_1|
|          1|       txn2|   1|               1_1|
|          1|       txn3|   2|               1_2|
|          2|       txn4|   2|               2_2|
|          3|       txn5|   0|               3_0|
+-----------+-----------+----+------------------+
 Um sicherzustellen, dass wir alle möglichen randomisierten Salzschlüssel in den großen Datensätzen abdecken, müssen wir Der kleine Datensatz mit allen möglichen Salzwerten explode // Step 2: Explode rows in smallDF for possible salted keys
val saltedSmallDF = (0 until numBuckets).toDF("salt").
    crossJoin(smallDF).
    withColumn("salted_customer_id", concat($"customer_id", lit("_"), $"salt")) 

saltedSmallDF.show()
+----+-----------+------+------------------+
|salt|customer_id|  name|salted_customer_id|
+----+-----------+------+------------------+
|   0|          1| Ahmed|               1_0|
|   1|          1| Ahmed|               1_1|
|   2|          1| Ahmed|               1_2|
|   0|          2|   Ali|               2_0|
|   1|          2|   Ali|               2_1|
|   2|          2|   Ali|               2_2|
|   0|          3|Hassan|               3_0|
|   1|          3|Hassan|               3_1|
|   2|          3|Hassan|               3_2|
+----+-----------+------+------------------+
 Jetzt können wir die beiden Datensätze leicht miteinander verbinden // Step 3: Perform salted join
val joinedDF = saltedLargeDF.
    join(saltedSmallDF, Seq("salted_customer_id", "customer_id"), "inner").
    select("customer_id", "transaction", "name")

joinedDF.show()
+-----------+-----------+------+
|customer_id|transaction|  name|
+-----------+-----------+------+
|          1|       txn2| Ahmed|
|          1|       txn1| Ahmed|
|          1|       txn3| Ahmed|
|          2|       txn4|   Ali|
|          3|       txn5|Hassan|
+-----------+-----------+------+
 Ein Beispiel in Python from pyspark.sql import SparkSession
from pyspark.sql.functions import col, rand, lit, concat
from pyspark.sql.types import IntegerType

# Simulated large dataset with skew
largeDF = spark.createDataFrame([
    (1, "txn1"), (1, "txn2"), (1, "txn3"), (2, "txn4"), (3, "txn5")
], ["customer_id", "transaction"])

# Small dataset
smallDF = spark.createDataFrame([
    (1, "Ahmed"), (2, "Ali"), (3, "Hassan")
], ["customer_id", "name"])

# Step 1: create a salting key in the large dataset
numBuckets = 3
saltedLargeDF = largeDF.withColumn("salt", (rand() * numBuckets).cast(IntegerType())) \
    .withColumn("salted_customer_id", concat(col("customer_id"), lit("_"), col("salt")))

# Step 2: Explode rows in smallDF for possible salted keys
salt_range = spark.range(0, numBuckets).withColumnRenamed("id", "salt")
saltedSmallDF = salt_range.crossJoin(smallDF) \
    .withColumn("salted_customer_id", concat(col("customer_id"), lit("_"), col("salt")))

# Step 3: Perform salted join
joinedDF = saltedLargeDF.join(
    saltedSmallDF,
    on=["salted_customer_id", "customer_id"],
    how="inner"
).select("customer_id", "transaction", "name")
 Notizen 
 
 
 
 Dieser Code verwendet spark.range(...) um Scala (0 bis numBuckets).toDF („Salz“) zu imitieren. Spalteausdrücke werden mithilfe von col (...), lit (...), und concat (...). Der Cast to integer verwendet .cast (IntegerType()). Tuning Tipp: Wählen numBuckets Numberspiele 
 
 
 Wenn Sie numBuckets = 100 festlegen, kann jeder Schlüssel in 100 Unterpartitionen aufgeteilt werden. Seien Sie jedoch vorsichtig, da die Verwendung von zu vielen Buckets die Leistung verringern kann, insbesondere bei Schlüsseln mit wenig Daten. If you know how to identify the skewed keys, then you can apply the salting for those keys only, and set the salting for other keys as literal  , e.x.
 
 0 
 
 
 
     // Step 1: create a salting key in the large dataset
        val numBuckets = 3
        val saltedLargeDF = largeDF.
            withColumn("salt", when($"customer_id" === 1, (rand() * numBuckets).cast("int")).otherwise(lit(0))).
            withColumn("salted_customer_id", concat($"customer_id", lit("_"), $"salt"))
  
    // Step 2: Explode rows in smallDF for possible salted keys
        val saltedSmallDF = (0 until numBuckets).toDF("salt").
            crossJoin(smallDF.filter($"customer_id" === 1)).
            select("customer_id", "salt", "name").
            union(smallDF.filter($"customer_id" =!= 1).withColumn("salt", lit(0)).select("customer_id", "salt", "name")).
            withColumn("salted_customer_id", concat($"customer_id", lit("_"), $"salt"))
 Beginnen Sie klein (z.B. 10-20) und erhöhen Sie allmählich auf der Grundlage der beobachteten Shuffle-Größen und der Aufgabenlaufzeit. Rule of Thumb: Endgültige Gedanken Salting ist eine effektive und einfache Methode zur Verwaltung von Abweichungen in Apache Spark, wenn traditionelle Partitionierung oder Hinweise ( Mit der richtigen Tuning und Überwachung kann diese Technik die Ausführungszeiten bei stark verzerrten Datensätzen deutlich reduzieren. SKEWED JOIN Ursprünglich veröffentlicht unter https://practical-software.com am 11. Mai 2025. Ursprünglich veröffentlicht in am 11. Mai 2025. https://practical-software.com https://practical-software.com https://practical-software.com

The code in this story is for educational purposes. The readers are solely responsible for whatever they build with it.

Walkthroughs, tutorials, guides, and tips. This story will teach you how to do something new or how to do something better.

So beheben Sie Data Skew in Apache Spark mit der Salting-Technik

About Author

KOMMENTARE

Hängeetiketten

DIESER ARTIKEL WURDE VORGESTELLT IN

Related Stories

HackerNoon Decoded 2024: Wir feiern unsere Gaming-Community!

HackerNoon's DeFi Writing Contest with SORA Network: XSTUSD Use Cases

HackerNoon Decoded 2024: Wir feiern die Community unserer Technologieunternehmen!

HackerNoon Decoded 2024: Wir feiern unsere Remote-Work-Community!

HackerNoon Decoded 2024: Wir feiern unsere Gaming-Community!

HackerNoon's DeFi Writing Contest with SORA Network: XSTUSD Use Cases

HackerNoon Decoded 2024: Wir feiern die Community unserer Technologieunternehmen!

HackerNoon Decoded 2024: Wir feiern unsere Remote-Work-Community!

Light-Mode

Classic

Newspaper

Dark-Mode

Neon Noir

Minty

HN StartUps