[jira] [Resolved] (SPARK-20309) Repartitioning - more than the default number of partitions

["Sean Owen (JIRA)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/SPARK-20309?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Sean Owen resolved SPARK-20309.
-------------------------------
    Resolution: Not A Problem

> Repartitioning - more than the default number of partitions
> -----------------------------------------------------------
>
>                 Key: SPARK-20309
>                 URL: https://issues.apache.org/jira/browse/SPARK-20309
>             Project: Spark
>          Issue Type: Bug
>          Components: Java API, SQL
>    Affects Versions: 1.6.1
>         Environment: Spark 1.6.1, Hadoop 2.6.2, Redhat Linux
>            Reporter: balaji krishnan
>
> We have a requirement to process roughly 45MM rows. Each row has a KEY column. The unique number of KEYS are roughly 25000. I do the following Spark SQL to get all 25000 partitions
> hiveContext.sql("select * from BigTbl distribute by KEY")
> This nicely creates all required partitions. Using MapPartitions we process all of these partitions in parallel. I have a variable which should be initiated once per partition. Lets call that variable name as designName. This variable is initialized with the KEY value (partition value) for future identification purposes.
>  FlatMapFunction<java.util.Iterator<Row>, DSGType> flatMapSetup = new FlatMapFunction<java.util.Iterator<Row>, DSGType>() {
>         String designName= null; 
>         @Override
>         public java.lang.Iterable<DSGType> call(java.util.Iterator<Row> it) throws Exception {
>             while (it.hasNext()) {
>                 Row row = it.next(); //row is of type org.apache.spark.sql.Row
>                 if (designName == null) { 
>     designName = row.getString(1); // row is of type org.apache.spark.sql.Row
> }
> .....
> }
> List<DSGType> Dsg =  partitionedRecs.mapPartitions(flatMapSetup).collect();
> The problem starts here for me.
> Since i have more than the default number of partitions (200) i do a re-partition to the unique number of partitions in my BigTbl. However if i re-partition the designName gets completely confused. Instead of processing a total of 25000 unique values, i get only 15000 values processed. For some reason the repartition completely messes up the uniqueness of the previous step. The statement that i use to 'distribute by' and subsequently repartition
> long distinctValues = hiveContext.sql("select KEY from BigTbl").distinct().count();
> JavaRDD<Row> partitionedRecs  = hiveContext.sql("select * from BigTbl DISTRIBUTE by SKU ").repartition((int)distinctValues).toJavaRDD().cache();
> I tried another solution, that is by changing the spark.sql.shuffle.partitions during SparkContext creation. Even that did not help. I get the same issue.
> new JavaSparkContext(new SparkConf().set("spark.driver.maxResultSize","0").set("spark.sql.shuffle.partitions","26000"))
> Is there a way to solve this issue please or is this a DISTRIBUTE BY bug ? We are using Spark 1.6.1
> Regards
> Bala



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org