Spark 2.0.2/2.1中DataFrame Join后Collect/Show阻塞问题求助

Hey there, let's dig into this Spark Join blocking issue you're facing when downgrading from Spark 2.2 to 2.0.2/2.1. I’ve got practical debugging steps and tuning tips tailored to your scenario:

First, fire up your Spark app and head to the Spark UI (default at http://localhost:4040). Focus on these key tabs to diagnose what’s stuck:

• Jobs/Stages: Locate the stage corresponding to your Join operation. Check if any tasks are stuck in RUNNING with zero progress, or if a small number of tasks are handling way more data than others (a clear sign of data skew).
• Storage: Look for excessive disk spills—if you see huge read/write numbers under "Spill (Disk)", your job is running out of memory to handle the Join, forcing data to disk and slowing things to a crawl.
• Executors: Monitor CPU and memory usage here. Keep an eye out for Out-of-Memory (OOM) errors in the logs, or frequent executor restarts—both can cause unexpected blocking.
• SQL Tab: Since you’re using DataFrames, this tab shows the execution plan for your Join. Compare it to the plan from Spark 2.2; you might find that Spark 2.0.x is using a less efficient Join strategy (like Shuffle Hash Join instead of Broadcast Hash Join) by default.

Spark 2.0.x has less aggressive auto-optimization than 2.2, so let’s tweak things manually:

1. Force Broadcast Hash Join for Small DataFrames
   If one of your DataFrames (say seasonalRatio) is small enough to fit in memory, explicitly use a broadcast join to avoid costly shuffles:

   from pyspark.sql.functions import broadcast
   freDataFrame = df_regressionLine.join(broadcast(seasonalRatio), condition)
   

   Just make sure the broadcasted DataFrame is under ~50MB (adjustable via spark.sql.autoBroadcastJoinThreshold) to avoid overwhelming executors.

2. Fix Data Skew
   Your Join uses four key fields—check if any combination of these fields has extremely uneven data distribution (e.g., one location_number accounts for 90% of records):

   # Check skew in df_regressionLine
   df_regressionLine.groupBy('location_number', 'location_type', 'pag_code', 'PERIOD')\
       .count().orderBy('count', ascending=False).show(20)
   # Check skew in seasonalRatio
   seasonalRatio.groupBy('location_number', 'location_type', 'pag_code', 'period')\
       .count().orderBy('count', ascending=False).show(20)
   

   If you find skew, split the problematic key into smaller subsets (e.g., add a random suffix to the skewed key), Join each subset separately, then combine the results.

3. Adjust Shuffle & Memory Parameters
   Tweak these Spark configs to match your data size:

   • spark.sql.shuffle.partitions: Default is 200. If you have large datasets, increase this to 500-1000 to split shuffle tasks into smaller chunks. For small datasets, decrease it to reduce task overhead.
   • spark.executor.memory: Bump this up (e.g., from 1g to 4g/8g, depending on your machine) to reduce disk spills during the Join.
   • spark.sql.autoBroadcastJoinThreshold: Raise this value (e.g., set to 52428800 for 50MB) if you want Spark to auto-broadcast larger DataFrames.

4. Verify Field Type Consistency
   Notice your Join uses df_regressionLine['PERIOD'] == seasonalRatio['period'] (case mismatch in field names). Double-check that these two fields have identical data types (e.g., both StringType or IntegerType). Implicit type conversions can cause unexpected overhead or data mismatches that lead to blocking.

Since you’re writing to Cassandra, don’t overlook connector compatibility:

• Ensure your connector version matches Spark 2.0.2/2.1: Spark 2.0 uses connector 1.6.x, Spark 2.1 uses connector 2.0.x. Using a connector built for Spark 2.2 can cause subtle issues.
• Avoid unnecessary collect() or show() calls before writing to Cassandra. Use the DataFrame writer directly to keep data distributed across executors:

  freDataFrame.write.format("org.apache.spark.sql.cassandra")\
      .options(table="your_table", keyspace="your_keyspace")\
      .save()
  

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