title: Normalize SPARK RDD partitions using reduceByKey(numPartitions) or repartition
body: <p>Using Spark 2.4.0. 
  My production data is extremely skewed, so one of the tasks was taking 7x longer than everything else. 
  I tried different strategies to normalize the data so that all executors worked equally - </p>
  
  <ol>
  <li>spark.default.parallelism </li>
  <li>reduceByKey(numPartitions)</li>
  <li>repartition(numPartitions) </li>
  </ol>
  
  <p>My expectation was that all three of them should evenly partition, however playing with some dummy non-production data on Spark Local/Standalone suggests that options 1,2 normalize better than 3. </p>
  
  <p>Data as below : (and i am trying to do a simple reduce on balance per account+ccy combination</p>
  
  <pre><code>account}date}ccy}amount
  A1}2020/01/20}USD}100.12
  A2}2010/01/20}SGD}200.24
  A2}2010/01/20}USD}300.36
  A1}2020/01/20}USD}400.12
  </code></pre>
  
  <p>Expected result should be <code>[A1-USD,500.24], [A2-SGD,200.24], [A2-USD,300.36]</code> Ideally these should be partitioned in 3 different partitions.</p>
  
  <pre><code>javaRDDWithoutHeader
  .mapToPair((PairFunction&lt;Balance, String, Integer&gt;) balance -&gt; new Tuple2&lt;&gt;(balance.getAccount() + balance.getCcy(), 1))        
      .mapToPair(new MyPairFunction())
     .reduceByKey(new ReductionFunction())
  </code></pre>
  
  <p>Code to check partitions </p>
  
  <pre><code>     System.out.println("b4 = " +pairRDD.getNumPartitions());
       System.out.println(pairRDD.glom().collect());
       JavaPairRDD&lt;DummyString, BigDecimal&gt; newPairRDD = pairRDD.repartition(3);
       System.out.println("Number of partitions = " +newPairRDD.getNumPartitions());
       System.out.println(newPairRDD.glom().collect());
  </code></pre>
  
  <ol>
  <li>Option 1: Doing nothing  </li>
  <li>Option 2: Setting spark.default.parallelism to 3 </li>
  <li>Option 3: reduceByKey with numPartitions = 3    </li>
  <li><p>Option 4: repartition(3)</p>
  
  <p>For Option 1
  Number of partitions = 2
  [
  [(DummyString{account='A2', ccy='SGD'},200.24), (DummyString{
  account='A2', ccy='USD'},300.36)], 
  [(DummyString{account='A1', ccy='USD'},500.24)]
  ]</p>
  
  <p>For option 2</p>
  
  <p>Number of partitions = 3
  [
  [(DummyString{account='A1', ccy='USD'},500.24)], 
  [(DummyString{account='A2', ccy='USD'},300.36)],
  [(DummyString{account='A2', ccy='SGD'},200.24)]]</p>
  
  <p>For option 3 
  Number of partitions = 3
  [
  [(DummyString{account='A1', ccy='USD'},500.24)],
  [(DummyString{account='A2', ccy='USD'},300.36)], 
  [(DummyString{account='A2', ccy='SGD'},200.24)]
  ]</p>
  
  <p>For option 4
  Number of partitions = 3
  [[], [(DummyString{
  account='A2', ccy='SGD'},200.24)], [(DummyString{
  account='A2', ccy='USD'},300.36), (DummyString{
  account='A1', ccy='USD'},500.24)]]</p></li>
  </ol>
  
  <p><strong>Conclusion :</strong> options 2(spark.default.parallelism) and 3(reduceByKey(numPartitions) normalized much better than option 4 (repartition)
  Fairly deterministic results, never saw option4 normalize into 3 partitions. </p>
  
  <p><strong>Question :</strong> </p>
  
  <ol>
  <li>is reduceByKey(numPartitions) much better than repartition or </li>
  <li>is this just because the sample data set is so small ? or </li>
  <li>is this behavior going to be different when we submit via a YARN cluster</li>
  </ol>