title: How to effectively join large tables in SparkSql?
body: <p>I am trying to improve performance on a  join involving two large tables using SparkSql. From various sources, I figured that the RDDs need to be partitioned.</p>
  
  <p>Source: <a href="https://deepsense.io/optimize-spark-with-distribute-by-and-cluster-by" rel="nofollow noreferrer">https://deepsense.io/optimize-spark-with-distribute-by-and-cluster-by</a></p>
  
  <p>However, when you load a file directly from a parquet file as given below, I am not sure how it can be created as a paired RDD!</p>
  
  <p>With Spark 2.0.1, using “cluster by” has no effect. </p>
  
  <pre><code>val rawDf1 = spark.read.parquet(“file in hdfs”)
  rawDf1 .createOrReplaceTempView(“rawdf1”)
  
  val rawDf2 = spark.read.parquet(“file in hdfs”)
  rawDf2 .createOrReplaceTempView(“rawdf2”)
  
  val rawDf3 = spark.read.parquet(“file in hdfs”)
  rawDf3 .createOrReplaceTempView(“rawdf3”)
  
  val df1 = spark.sql(“select * from rawdf1 cluster by key)
  df1 .createOrReplaceTempView(“df1”)
  
  val df2 = spark.sql(“select * from rawdf2 cluster by key)
  df2 .createOrReplaceTempView(“df2”)
  
  val df3 = spark.sql(“select * from rawdf3 cluster by key)
  df3 .createOrReplaceTempView(“df3”)
  
  val resultDf = spark.sql(“select * from df1 a inner join df2 b on a.key = b.key inner join df3 c on a.key =c.key”)
  </code></pre>
  
  <p>Whether I use "cluster by" key or not, I still see the same query plan being generated by Spark. How can I create a rdd pair in spark sql so that joins can use tables that can be partitioned?</p>
  
  <p>Without proper partitioning, a lot of shuffles are happening resulting in long delays.</p>
  
  <p>Our configuration ( 5 worker nodes with 1 executor (5 cores per executor) each having 32 cores and 128 GB of RAM): </p>
  
  <pre><code>spark.cores.max 25
  spark.default.parallelism   75
  spark.driver.extraJavaOptions   -XX:+UseG1GC
  spark.executor.memory   60G
  spark.rdd.compress  True
  spark.driver.maxResultSize  4g
  spark.driver.memory 8g
  spark.executor.cores    5
  spark.executor.extraJavaOptions -Djdk.nio.maxCachedBufferSize=262144
  spark.memory.storageFraction    0.2
  </code></pre>
  
  <p>To add more info: I am joining more than one table in the same select using the same key across all tables. So it is not possible to create a dataframe first to call repartitionby. I understand I can do this using dataframe api. But my question is how to accomplish this using plain sparksql.</p>