[Spark SQL]: how is “Exchange hashpartitioning” working in spark

classic Classic list List threaded Threaded
1 message Options
nkx
Reply | Threaded
Open this post in threaded view
|

[Spark SQL]: how is “Exchange hashpartitioning” working in spark

nkx
Hi,

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.

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?


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.

I have tested this environment on
- SPARK 2.0 on cloudera CDH 5.9.3
- SPARK 2.3.1 on emr-5.17.0


Here is my tests with outputs. Please use Spark-shell to run them


        scala> import org.apache.spark.HashPartitioner
        import org.apache.spark.HashPartitioner

        scala> val mydataindex = Array(0,1, 2, 3,4)
        mydataindex: Array[Int] = Array(0, 1, 2, 3, 4)

        scala> val mydata = sc.parallelize(for {
             |  x <- mydataindex
             |  y <- Array(123,456,789)
             | } yield (x, y), 100)
        mydata: org.apache.spark.rdd.RDD[(Int, Int)] =
ParallelCollectionRDD[0] at parallelize at <console>:27

        scala> val rddMyData = mydata.partitionBy(new HashPartitioner(5))
        rddMyData: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[1] at
partitionBy at <console>:26

        scala> val rddMyDataPartitions =   rddMyData.mapPartitionsWithIndex{
             |                 (index, iterator) => {
             |                    val myList = iterator.toList
             |                    myList.map(x => x + " -> " +
index).iterator
             |                 }
             |              }
        rddMyDataPartitions: org.apache.spark.rdd.RDD[String] =
MapPartitionsRDD[2] at mapPartitionsWithIndex at <console>:26

        scala>
             | // this is expected:

        scala> rddMyDataPartitions.take(100)
        res1: Array[String] = Array((0,123) -> 0, (0,456) -> 0, (0,789) ->
0, (1,123) -> 1, (1,456) -> 1, (1,789) -> 1, (2,123) -> 2, (2,456) -> 2,
(2,789) -> 2, (3,456) -> 3, (3,789) -> 3, (3,123) -> 3, (4,789) -> 4,
(4,123) -> 4, (4,456) -> 4)

        scala> val dfMyData = mydata.toDF()
        dfMyData: org.apache.spark.sql.DataFrame = [_1: int, _2: int]

        scala> val dfMyDataRepartitioned = dfMyData.repartition(5,col("_1"))
        dfMyDataRepartitioned:
org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [_1: int, _2: int]

        scala> 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> 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> // this is unexpected, because 1 partition has more indexes

        scala> dfMyDataRepartitionedPartition.show()
        +------------+-----+
        |partition_id|count|
        +------------+-----+
        |           1|    6|
        |           3|    3|
        |           4|    3|
        |           2|    3|
        +------------+-----+



I also wrote this question to stackoverflow, but I wanted to connect the
experts directly as well :
https://stackoverflow.com/questions/54215601/how-is-exchange-hashpartitioning-working-in-spark


Thanks in advance!




--
Sent from: http://apache-spark-user-list.1001560.n3.nabble.com/

---------------------------------------------------------------------
To unsubscribe e-mail: [hidden email]