PySpark大表关联数据倾斜消除：含大量NULL值表的处理方法

Let’s break down exactly how to resolve this skew issue—since you’ve got a massive Table A (150M rows) with tons of NULLs in column_1 joining a tiny Table B (60 rows), we can target both tables to spread the workload evenly across executors.

Why the Skew Happens #

All NULL values in column_1 hash to the same value, so Spark sends every single one of those rows to the same executor. When you run a LEFT JOIN with Table B, that executor ends up handling all the NULL rows plus the full join with Table B’s 60 rows—way more work than any other executor can handle.

Steps for Table A #

We need to "salt" the NULL values in column_1 to split them across multiple partitions. Here’s how:

• Add a random suffix to NULL values: Create a new column (salted_column_1) where NULLs are replaced with a random number (e.g., 0 to 9), and non-NULL values stay unchanged. This splits the NULL rows into 10 distinct partitions instead of one.

  // Scala example
  import org.apache.spark.sql.functions.{when, rand, floor}
  
  val saltedTableA = tableA.withColumn(
    "salted_column_1",
    when(col("column_1").isNull, floor(rand() * 10).cast("string"))
      .otherwise(col("column_1"))
  )
  

  # PySpark example
  from pyspark.sql.functions import when, rand, floor
  
  salted_table_a = table_a.withColumn(
      "salted_column_1",
      when(table_a.column_1.isNull, floor(rand() * 10).cast("string"))
      .otherwise(table_a.column_1)
  )
  

  Adjust the number 10 based on your skew severity—use a higher number if you have an extreme number of NULL rows (e.g., 20 or 30 for 100M+ NULLs).

Steps for Table B #

Since Table B is tiny, we can replicate it to match the salt values we added to Table A. This ensures each salted NULL partition from Table A can join with a copy of Table B:

• Cross join with a salt list: Create a small DataFrame containing the same salt values (0 to 9) and cross join it with Table B. This creates 10 copies of Table B, each tagged with a unique salt.

  // Scala example
  val saltList = spark.range(0, 10).select(col("id").cast("string").alias("salt"))
  val replicatedTableB = tableB.crossJoin(saltList)
                                .withColumn(
                                  "salted_column_1",
                                  when(col("column_1").isNull, col("salt"))
                                    .otherwise(col("column_1"))
                                )
                                .drop("salt")
  

  # PySpark example
  salt_list = spark.range(0, 10).selectExpr("cast(id as string) as salt")
  replicated_table_b = table_b.crossJoin(salt_list) \
                              .withColumn(
                                  "salted_column_1",
                                  when(table_b.column_1.isNull, salt_list.salt)
                                  .otherwise(table_b.column_1)
                              ) \
                              .drop("salt")
  

  If Table B has no NULLs in column_1, the when clause won’t modify anything—we just ensure the salted join key matches Table A’s structure.

Perform the JOIN #

Now join using the salted_column_1 instead of the original column_1, then clean up the salted column:

// Scala
val joinedDF = saltedTableA.join(replicatedTableB, Seq("salted_column_1"), "left")
                           .drop("salted_column_1")

# PySpark
joined_df = salted_table_a.join(replicated_table_b, on="salted_column_1", how="left") \
                          .drop("salted_column_1")

Bonus: Spark 3.0+ Alternative #

If you’re on Spark 3.0 or newer, you can enable automatic skew handling with:

SET spark.sql.adaptive.skewJoin.enabled = true;

But this automatic approach might not handle NULL-driven skew as reliably as manual salting, especially with your extreme NULL volume. Manual salting gives you full control over how the load is distributed.

Key Notes #

• Since Table B is only 60 rows, replicating it 10 times results in just 600 rows—no meaningful overhead here.
• This method works across most Spark versions (2.x and 3.x), so you don’t have to worry about version-specific conflicts.

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