Lorsque vous travaillez avec de grands ensembles de données dans Un problème de performance est Cela se produit lorsque quelques clés la distribution des données, conduisant à partitions et des requêtes lentes. Cela se produit principalement pendant les opérations qui nécessitent comme Ou même régulier . Apache Spark data skew dominate uneven shuffling joins aggregations Une méthode pratique pour réduire les déchets est , qui implique la diffusion artificielle de clés lourdes sur plusieurs partitions. Dans ce post, je vous guiderai à travers cela avec un exemple pratique. salting Comment Salting résout les problèmes de détournement de données En ajoutant a généré le nombre à la clé jointe, puis en se joignant sur cette clé combinée, nous pouvons distribuer les grandes clés plus uniformément.Cela rend la distribution des données plus uniforme et répartit la charge sur plus de travailleurs, au lieu d'envoyer la plupart des données à un travailleur et de laisser les autres inactives. randomly Les bienfaits du salage 
 
 
 
 
 
  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: Quand utiliser le saumon Lors de joints ou d'agrégations avec des clés déformées, utilisez le saltage lorsque vous remarquez de longs temps de déformation ou des erreurs d'exécuteur en raison d'une déformation des données. Il est également utile dans les applications de streaming en temps réel où la partitionnement affecte l'efficacité du traitement des données, ou lorsque la plupart des travailleurs sont inactives alors que quelques-uns sont coincés dans un état d'exécution. L’exemple de l’escalade Nous allons générer quelques données avec un Nous pouvons supposer qu'il y a deux ensembles de données que nous devons joindre: l'un est un grand ensemble de données, et l'autre est un petit ensemble de données. 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")
 Ajoutez la colonne salée aux grands ensembles de données, que nous utilisons pour distribuer les valeurs de la grande clé dans des partitions plus petites 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|
+-----------+-----------+----+------------------+
 Pour nous assurer que nous couvrons toutes les clés salées aléatoires possibles dans les grands ensembles de données, nous devons le petit ensemble de données avec toutes les valeurs salées possibles 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|
+----+-----------+------+------------------+
 Maintenant, nous pouvons facilement joindre les deux ensembles de données // 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|
+-----------+-----------+------+
 Exemple en 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")
 Notes 
 
 
 
 Ce code utilise spark.range(...) pour imiter Scala (0 à numBuckets).toDF (« sel »). Les expressions de colonne sont traitées en utilisant col (...), lit (...), et concat (...). Le cast à integer utilise .cast (IntegerType()). Tuning : le choix numBuckets Numérotation 
 
 
 Si vous définissez numBuckets = 100, chaque clé peut être divisée en 100 sous-partitions. Cependant, soyez prudent car l'utilisation de trop de buckets peut diminuer la performance, en particulier pour les clés avec peu de données. 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"))
 Commencez petit (par exemple, 10-20) et augmentez progressivement en fonction des tailles de shuffle observées et du temps d'exécution des tâches. Rule of Thumb: Pensées finales Le saltage est une méthode efficace et simple pour gérer la déviation dans Apache Spark lorsqu'il s'agit de partitions traditionnelles ou d'indices ( Avec le bon réglage et la surveillance, cette technique peut réduire considérablement les temps d'exécution des tâches sur des ensembles de données très déformés. SKEWED JOIN Publié à https://practical-software.com le 11 mai 2025. Publié à l'origine à Le 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.

Read My Stories

Cet audio est produit dans la langue originale de l'histoire !

Trop long; Pour lire

Make resilience your competitive advantage

Comment résoudre le détournement de données dans Apache Spark avec la technique Salting

About Author

COMMENTAIRES

ÉTIQUETTES

CET ARTICLE A ÉTÉ PARU DANS

Related Stories

Libérer la puissance de l’IA. Une revue systématique des techniques de pointe : résumé et introduction

Des forums aux flux : comment les algorithmes des réseaux sociaux façonnent l'interaction numérique

Les couches invisibles : pourquoi les entretiens avec les utilisateurs sont un atout irremplaçable

Le modèle Bitcoin UTXO, alimentant un écosystème unique

Libérer la puissance de l’IA. Une revue systématique des techniques de pointe : résumé et introduction

Des forums aux flux : comment les algorithmes des réseaux sociaux façonnent l'interaction numérique

Les couches invisibles : pourquoi les entretiens avec les utilisateurs sont un atout irremplaçable

Le modèle Bitcoin UTXO, alimentant un écosystème unique

Light-Mode

Classic

Newspaper

Dark-Mode

Neon Noir

Minty

HN StartUps