Spark SQL(Scala)：如何高效连接两张存在键倾斜的超大表？

Hey there! Let's work through this skewed join challenge you're dealing with in Spark 2.2—those 1B and 5B row tables are no joke, and it's smart you already split each into two DataFrames. I’m guessing you separated the skewed high-frequency keys from the rest of the evenly distributed ones? That’s the perfect foundation. Here’s how to wrap up the ETL to handle the skew efficiently, no Hive config changes needed:

First, double-check that your split correctly isolates the problematic keys. You can identify skewed keys by running a quick count on the join column:

// Run this on one of your tables to find top skewed keys
val keyCounts = spark.sql("SELECT join_key, COUNT(*) as cnt FROM your_table GROUP BY join_key ORDER BY cnt DESC LIMIT 10")
keyCounts.show()

Use these top keys to split each table into two DataFrames:

• nonSkewedDF1 / nonSkewedDF2: All rows with evenly distributed join keys
• skewedDF1 / skewedDF2: Only rows with the high-frequency skewed join keys

This is the easy part—since the keys here are evenly distributed, a standard join won’t cause task bottlenecks. Keep it simple:

// Inner join example; adjust join type to match your needs (left/right/full)
val nonSkewedResult = nonSkewedDF1.join(nonSkewedDF2, Seq("join_key"), "inner")

This is where you’ll fix the skew. Choose one of these methods based on how many skewed keys you have:

Option A: Salting (Best for Moderate to High Skew) #

Salting breaks up the large skewed partitions by adding a random suffix to the join key in the larger table, then expanding the smaller table’s skewed rows to match all possible suffixes. Here’s how to implement it:

import org.apache.spark.sql.functions._

// Pick a salt count based on skew severity (e.g., 100 for extremely skewed keys)
val saltCount = 100

// Add salt to the larger table's skewed subset
val skewedDF2Salted = skewedDF2.withColumn(
  "salted_join_key",
  concat(col("join_key"), lit("_"), floor(rand() * saltCount))
)

// Expand the smaller table's skewed subset to include all salt values
val skewedDF1Expanded = skewedDF1.withColumn(
  "salt",
  explode(array((0 until saltCount).map(lit(_)): _*))
).withColumn(
  "salted_join_key",
  concat(col("join_key"), lit("_"), col("salt"))
).drop("salt")

// Join the salted subsets, then clean up the salted key
val skewedResult = skewedDF1Expanded.join(skewedDF2Salted, Seq("salted_join_key"), "inner")
  .drop("salted_join_key")

Option B: Per-Key Joins (Best for a Small Number of Skewed Keys) #

If you only have a handful of skewed keys (e.g., 5-10), process each one individually to avoid overloading any single task:

// List of your identified skewed keys
val skewedKeys = List("high_skew_key_1", "high_skew_key_2", "high_skew_key_3")

// Join each skewed key separately, then union the results
val skewedResults = skewedKeys.map(key => {
  val filteredDF1 = skewedDF1.filter(col("join_key") === key)
  val filteredDF2 = skewedDF2.filter(col("join_key") === key)
  filteredDF1.join(filteredDF2, Seq("join_key"), "inner")
})

val skewedResult = skewedResults.reduce(_ union _)

Finally, union the non-skewed and skewed join outputs to get your full result set:

val finalETLResult = nonSkewedResult.union(skewedResult)

• Tune spark.sql.shuffle.partitions: Set this to a higher value (e.g., 2000-5000) based on your cluster’s core count—this prevents shuffle partitions from being too large.
• Enable adaptive execution: Spark 2.2 supports adaptive query execution, which automatically adjusts partition sizes during runtime. Turn it on with:

  spark.conf.set("spark.sql.adaptive.enabled", "true")
  

• Use partition pruning: If your Hive tables are partitioned, make sure your initial queries filter on partition columns to reduce the data you’re loading into Spark.

内容的提问来源于stack exchange，提问作者Guigs