Got it, let's tackle this Spark SQL performance issue with your 50M-record dataset—20 minutes for full processing with inner/left joins is definitely something we can trim down. Here are practical, actionable steps I've used in similar production scenarios:

• Partition your tables strategically: Pick a column you frequently filter on (like a date dt or business ID) to partition your data. This lets Spark skip scanning irrelevant partitions entirely instead of reading the full dataset. For example, when writing data:

  df.write.partitionBy("dt").parquet("/path/to/partitioned-data")
  

  Spark automatically applies partition pruning when your query includes a filter on the partition column.
• Bucket for frequent joins: If you're joining on a fixed key (like user_id), bucket the involved tables by that key. Aim for a bucket count that's 2-3x your cluster's total core count—this eliminates full shuffles during joins by matching buckets directly. Example:

  df.write.bucketBy(100, "user_id").sortBy("user_id").saveAsTable("bucketed_user_table")
  

• Switch to columnar storage: Ditch CSV/JSON for Parquet or ORC. These formats are compressed, columnar, and support predicate pushdown—meaning Spark can skip reading entire columns you don't need. Enable Snappy compression for a great balance of speed and space:

  df.write.parquet("/path/to/parquet-data", compression="snappy")
  

• Broadcast small tables: For inner/left joins where one table is significantly smaller (say, <10GB), use the broadcast hint to send the small table to all executors. This avoids shuffling the large table entirely. Example in SQL:

  SELECT * FROM large_table L 
  JOIN broadcast(small_reference_table) S 
  ON L.user_id = S.user_id
  

  Spark might do this automatically, but explicitly adding the hint ensures it uses this optimal strategy.
• Fix data skew: If a single join key (like NULL or a super-popular ID) is causing one executor to hog all the work, try these fixes:
  • Filter out NULL join keys if your business logic allows it—NULLs often cause skew.
  • Add a random prefix to skewed keys, split the join into smaller chunks, then recombine results.
  • Enable adaptive skew join optimization (Spark 3.0+): spark.sql.adaptive.skewJoin.enabled=true—this handles skew automatically in many cases.
• Reorder joins manually (if needed): Spark's optimizer usually picks the best order, but if you know a small table should be joined first to reduce dataset size early, structure your query to do that.

• Right-size your executors: Avoid overloading or underutilizing executors. A good starting point is 4-8 cores per executor, with 16-32GB of memory (reserve ~20% for off-heap memory). Adjust these parameters:
  • spark.executor.cores
  • spark.executor.memory
  • spark.driver.memory (make sure it's enough to handle metadata)
• Adjust shuffle partitions: The default 200 partitions is often too low for 50M records. Increase it to 1000-2000 (match to your cluster's capacity) with spark.sql.shuffle.partitions=1500. This distributes work more evenly across executors.
• Enable Adaptive Query Execution (AQE): Spark 3.0+ game-changer. AQE automatically adjusts shuffle partitions, handles skew, and optimizes join strategies on the fly. Turn it on with:

  spark.sql.adaptive.enabled=true
  

• Stop using SELECT *: Only select the columns you actually need. This reduces data transfer across the network and cuts down on memory usage.
• Filter early: Apply WHERE clauses before joining tables, not after. This reduces the amount of data that gets shuffled during joins. For example:

  -- Good: Filter first, then join
  SELECT L.id, S.score 
  FROM (SELECT id, user_id FROM large_table WHERE dt >= '2024-01-01') L
  JOIN small_table S ON L.user_id = S.user_id
  
  -- Bad: Join first, then filter
  SELECT L.id, S.score 
  FROM large_table L
  JOIN small_table S ON L.user_id = S.user_id
  WHERE L.dt >= '2024-01-01'
  

• Use CTEs for repeated logic: If you're reusing a subquery multiple times, wrap it in a CTE to avoid redundant computation:

  WITH filtered_large_data AS (
    SELECT id, user_id FROM large_table WHERE dt >= '2024-01-01'
  )
  SELECT * FROM filtered_large_data JOIN small_table S ON filtered_large_data.user_id = S.user_id
  

• If you run this query multiple times, or if intermediate results are used across queries, cache them with CACHE TABLE or persist. For example:

  CACHE TABLE filtered_large_data;
  

  Just remember to uncache it when you're done to free up resources: UNCACHE TABLE filtered_large_data;

Start with the low-effort, high-impact wins first—like enabling AQE, switching to Parquet, and checking for data skew. Then move to partitioning/bucketing if those don't give you enough speedup.

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