@Omri thanks for your question!

To help optimize your complex join further, we need clarification on a few details:

• Data Characteristics:

  • Approximate size of df1 and df2 (in rows and/or size).
  • Distribution of main_key in both dataframes—are the top N keys dominating the distribution, and how many keys are significantly skewed?

• Cluster Configuration:

  • Details of your cluster (e.g., number of nodes, cores, memory per executor).
  • Are you using spot instances or dedicated nodes?

• Data Source:

  • What is the data source for df1 and df2? (e.g., Delta tables, Parquet, or external sources like Redshift/S3?)
  • Are the datasets partitioned, and if so, how are they partitioned?

• Query Execution Insights:

  • Have you examined the Spark UI or execution plans? What stages or tasks have the most skew?
  • Are you seeing significant shuffle writes or long-running tasks for specific partitions?

• Previous Attempts:

  • When salting or exploding, what specific methods did you use? Did you adjust the degree of salting or the granularity of exploding?
  • Did you test alternative methods for handling skewed keys, like key bucketing or additional partitioning?
  • Were there cases where broadcasting or exploding significantly reduced execution time, or did skew persist across all approaches?

If you could additionally capture some (traceback) python stacktraces, we can get visibility on the operation that is slow running, and potentially correlate with the Spark UI metrics for the slow SQL ID and Stage.