title: Apache spark join with dynamic re-partitionion
body: <p>I'm trying to do a fairly straightforward join on two tables, nothing complicated. Load both tables, do a join and update columns but it keeps throwing an exception.</p>
  <p>I noticed the task is stuck on the last partition <code>199/200</code> and eventually crashes. My suspicion is that the data is skewed causing all the data to be loaded in the last partition <code>199</code>.</p>
  <p><code>SELECT COUNT(DISTINCT report_audit) FROM ReportDs = 1.5million.</code></p>
  <p>While</p>
  <p><code>SELECT COUNT(*) FROM ReportDs = 57million.</code></p>
  <p>Cluster details: CPU: 40 cores, Memory: 160G.</p>
  <p>Here is my sample code:</p>
  <pre><code>...
  def main(args: Array[String]) {
  
    val log = LogManager.getRootLogger
    log.setLevel(Level.INFO)
  
    val conf = new SparkConf().setAppName(&quot;ExampleJob&quot;)
                            //.setMaster(&quot;local[*]&quot;)
                            //.set(&quot;spark.sql.shuffle.partitions&quot;, &quot;3000&quot;)
                            //.set(&quot;spark.sql.crossJoin.enabled&quot;, &quot;true&quot;)
                            .set(&quot;spark.storage.memoryFraction&quot;, &quot;0.02&quot;)
                            .set(&quot;spark.shuffle.memoryFraction&quot;, &quot;0.8&quot;)
                            .set(&quot;spark.serializer&quot;, &quot;org.apache.spark.serializer.KryoSerializer&quot;)
                            .set(&quot;spark.default.parallelism&quot;, (CPU * 3).toString)
  
  
    val sparkSession = SparkSession.builder()
                                   .config(conf)
                                   .getOrCreate()
  
  
    val reportOpts = Map(
                &quot;url&quot;     -&gt; s&quot;jdbc:postgresql://$DB_HOST:$DB_PORT/$DATABASE&quot;,
                &quot;driver&quot;  -&gt; &quot;org.postgresql.Driver&quot;,
                &quot;dbtable&quot; -&gt; &quot;REPORT_TBL&quot;,
                &quot;user&quot;    -&gt; DB_USER,
                &quot;password&quot;-&gt; DB_PASSWORD,
                &quot;partitionColumn&quot; -&gt; RPT_NUM_PARTITION,
                &quot;lowerBound&quot; -&gt; RPT_LOWER_BOUND,
                &quot;upperBound&quot; -&gt; RPT_UPPER_BOUND,
                &quot;numPartitions&quot; -&gt; &quot;200&quot;
              )
  
  
    val accountOpts = Map(
                  &quot;url&quot;     -&gt; s&quot;jdbc:postgresql://$DB_HOST:$DB_PORT/$DATABASE&quot;,
                  &quot;driver&quot;  -&gt; &quot;org.postgresql.Driver&quot;,
                  &quot;dbtable&quot; -&gt; ACCOUNT_TBL,
                  &quot;user&quot;    -&gt; DB_USER,
                  &quot;password&quot;-&gt; DB_PASSWORD,
                  &quot;partitionColumn&quot; -&gt; ACCT_NUM_PARTITION,
                  &quot;lowerBound&quot; -&gt; ACCT_LOWER_BOUND,
                  &quot;upperBound&quot; -&gt; ACCT_UPPER_BOUND,
                  &quot;numPartitions&quot; -&gt; &quot;200&quot;
                )
  
    val sc = sparkSession.sparkContext;
  
    import sparkSession.implicits._
  
    val reportDs = sparkSession.read.format(&quot;jdbc&quot;).options(reportOpts).load.cache().alias(&quot;a&quot;)
  
    val accountDs = sparkSession.read.format(&quot;jdbc&quot;).options(accountOpts).load.cache().alias(&quot;c&quot;)
  
    val reportData =  reportDs.join(accountDs, reportDs(&quot;report_audit&quot;) === accountDs(&quot;reference_id&quot;))
                                          .withColumn(&quot;report_name&quot;, when($&quot;report_id&quot; === &quot;xxxx-xxx-asd&quot;, $&quot;report_id_ref_1&quot;)
                                                                     .when($&quot;report_id&quot; === &quot;demoasd-asdad-asda&quot;, $&quot;report_id_ref_2&quot;)
                                                                     .otherwise($&quot;report_id_ref_1&quot; + &quot;:&quot; + $&quot;report_id_ref_2&quot;))
                                          .withColumn(&quot;report_version&quot;, when($&quot;report_id&quot; === &quot;xxxx-xxx-asd&quot;, $&quot;report_version_1&quot;)
                                                                         .when($&quot;report_id&quot; === &quot;demoasd-asdad-asda&quot;, $&quot;report_version_2&quot;)
                                                                         .otherwise($&quot;report_version_3&quot;))
                                          .withColumn(&quot;status&quot;, when($&quot;report_id&quot; === &quot;xxxx-xxx-asd&quot;, $&quot;report_status&quot;)
                                                                  .when($&quot;report_id&quot; === &quot;demoasd-asdad-asda&quot;, $&quot;report_status_1&quot;)
                                                                  .otherwise($&quot;report_id&quot;))
                                          .select(&quot;...&quot;)
  
  
  
  
  
  
    val prop = new Properties()
    prop.setProperty(&quot;user&quot;, DB_USER)
    prop.setProperty(&quot;password&quot;, DB_PASSWORD)
    prop.setProperty(&quot;driver&quot;, &quot;org.postgresql.Driver&quot;)
  
  
    reportData.write
                    .mode(SaveMode.Append)
                    .jdbc(s&quot;jdbc:postgresql://$DB_HOST:$DB_PORT/$DATABASE&quot;, &quot;cust_report_data&quot;, prop)
  
  
    sparkSession.stop()
  </code></pre>
  <p>I think there should be an elegant way to handle this sort of data skewness.</p>