title: how is "Exchange hashpartitioning" working in spark
body: <p>I have a dataset which I want to write sorted into parquet files for getting benefit of requesting these files afterwards over Spark including Predicate Pushdown.</p>
  
  <p>Currently I used repartition by column and the number of partitions to move the data to the particular partition. The column is identifying the corresponding partition (beginning from 0 to (fixed) n). The result is that scala/spark is generating an unexpected result and creating less partitions (some of them are empty). Maybe a Hash Collision? </p>
  
  <p>For solving the problem I tried to find out the reason and tried to find workarounds. I found one workaround by transforming the dataframe to rdd and  use partitionBy with HashPartitioner. Surprising for me: I got the expected results. But converting a dataframe to an RDD is not a solution for me, because it takes too much resources.</p>
  
  <p>I have tested this environment on </p>
  
  <ul>
  <li><p>SPARK 2.0 on cloudera CDH 5.9.3</p></li>
  <li><p>SPARK 2.3.1 on emr-5.17.0</p></li>
  </ul>
  
  <p>Here is my tests with outputs. Please use Spark-shell to run them</p>
  
  <pre><code>    scala&gt; import org.apache.spark.HashPartitioner
      import org.apache.spark.HashPartitioner
  
      scala&gt; val mydataindex = Array(0,1, 2, 3,4)
      mydataindex: Array[Int] = Array(0, 1, 2, 3, 4)
  
      scala&gt; val mydata = sc.parallelize(for {
           |  x &lt;- mydataindex
           |  y &lt;- Array(123,456,789)
           | } yield (x, y), 100)
      mydata: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[0] at parallelize at &lt;console&gt;:27
  
      scala&gt; val rddMyData = mydata.partitionBy(new HashPartitioner(5))
      rddMyData: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[1] at partitionBy at &lt;console&gt;:26
  
      scala&gt; val rddMyDataPartitions =   rddMyData.mapPartitionsWithIndex{
           |                 (index, iterator) =&gt; {
           |                    val myList = iterator.toList
           |                    myList.map(x =&gt; x + " -&gt; " + index).iterator
           |                 }
           |              }
      rddMyDataPartitions: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[2] at mapPartitionsWithIndex at &lt;console&gt;:26
  
      scala&gt;
           | // this is expected:
  
      scala&gt; rddMyDataPartitions.take(100)
      res1: Array[String] = Array((0,123) -&gt; 0, (0,456) -&gt; 0, (0,789) -&gt; 0, (1,123) -&gt; 1, (1,456) -&gt; 1, (1,789) -&gt; 1, (2,123) -&gt; 2, (2,456) -&gt; 2, (2,789) -&gt; 2, (3,456) -&gt; 3, (3,789) -&gt; 3, (3,123) -&gt; 3, (4,789) -&gt; 4, (4,123) -&gt; 4, (4,456) -&gt; 4)
  
      scala&gt; val dfMyData = mydata.toDF()
      dfMyData: org.apache.spark.sql.DataFrame = [_1: int, _2: int]
  
      scala&gt; val dfMyDataRepartitioned = dfMyData.repartition(5,col("_1"))
      dfMyDataRepartitioned: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [_1: int, _2: int]
  
      scala&gt; dfMyDataRepartitioned.explain(false)
      == Physical Plan ==
      Exchange hashpartitioning(_1#3, 5)
      +- *(1) SerializeFromObject [assertnotnull(input[0, scala.Tuple2, true])._1 AS _1#3, assertnotnull(input[0, scala.Tuple2, true])._2 AS _2#4]
         +- Scan ExternalRDDScan[obj#2]
  
      scala&gt; val dfMyDataRepartitionedPartition  = dfMyDataRepartitioned.withColumn("partition_id", spark_partition_id()).groupBy("partition_id").count()
      dfMyDataRepartitionedPartition: org.apache.spark.sql.DataFrame = [partition_id: int, count: bigint]
  
      scala&gt; // this is unexpected, because 1 partition has more indexes
  
      scala&gt; dfMyDataRepartitionedPartition.show()
      +------------+-----+
      |partition_id|count|
      +------------+-----+
      |           1|    6|
      |           3|    3|
      |           4|    3|
      |           2|    3|
      +------------+-----+
  </code></pre>
  
  <p>I first throught that HashPartitioner is been used in the method of repartitioning a dataframe, but this doesn't seem to be the case, becaue it is working on RDDs.</p>
  
  <p>Could anyone guide me how this "Exchange hashpartitioning" (see explain output above) is working?</p>
  
  <p>2019-01-16 12:20: This is not a duplicate of <a href="https://stackoverflow.com/questions/31424396">How does HashPartitioner work?</a> because I am interested in the Hashing Algorithm of repartition by column (+ number partitions) on a Integer column. The general HashPartitioner is working as expected as you can see in the source code.</p>