[
https://issues.apache.org/jira/browse/SPARK-24343?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Apache Spark reassigned SPARK-24343:
------------------------------------
Assignee: Apache Spark
> Avoid shuffle for the bucketed table when shuffle.partition > bucket number
> ---------------------------------------------------------------------------
>
> Key: SPARK-24343
> URL: https://issues.apache.org/jira/browse/SPARK-24343
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 2.3.0
> Reporter: yucai
> Assignee: Apache Spark
> Priority: Major
>
> When shuffle.partition > bucket number, Spark needs to shuffle the bucket
> table as per the shuffle.partition, can we avoid this?
> See below example:
> {code:java}
> CREATE TABLE dev
> USING PARQUET
> AS SELECT ws_item_sk, i_item_sk
> FROM web_sales_bucketed
> JOIN item ON ws_item_sk = i_item_sk;{code}
> Suppose web_sales_bucketed is bucketed into 4 buckets and set
> shuffle.partition=10.
> Currently, both tables are shuffled into 10 partitions.
> {code:java}
> Execute CreateDataSourceTableAsSelectCommand
> CreateDataSourceTableAsSelectCommand `dev`, ErrorIfExists, [ws_item_sk#2,
> i_item_sk#6]
> +- *(5) SortMergeJoin [ws_item_sk#2], [i_item_sk#6], Inner
> :- *(2) Sort [ws_item_sk#2 ASC NULLS FIRST], false, 0
> : +- Exchange hashpartitioning(ws_item_sk#2, 10)
> : +- *(1) Project [ws_item_sk#2]
> : +- *(1) Filter isnotnull(ws_item_sk#2)
> : +- *(1) FileScan parquet tpcds10.web_sales_bucketed[ws_item_sk#2] Batched:
> true, Format: Parquet, Location:...
> +- *(4) Sort [i_item_sk#6 ASC NULLS FIRST], false, 0
> +- Exchange hashpartitioning(i_item_sk#6, 10)
> +- *(3) Project [i_item_sk#6]
> +- *(3) Filter isnotnull(i_item_sk#6)
> +- *(3) FileScan parquet tpcds10.item[i_item_sk#6] Batched: true, Format:
> Parquet, Location:...
> {code}
> A better plan should avoid the shuffle in the bucket table.
> {code:java}
> Execute CreateDataSourceTableAsSelectCommand
> CreateDataSourceTableAsSelectCommand `dev`, ErrorIfExists, [ws_item_sk#2,
> i_item_sk#6]
> +- *(4) SortMergeJoin [ws_item_sk#2], [i_item_sk#6], Inner
> :- *(1) Sort [ws_item_sk#2 ASC NULLS FIRST], false, 0
> : +- *(1) Project [ws_item_sk#2]
> : +- *(1) Filter isnotnull(ws_item_sk#2)
> : +- *(1) FileScan parquet tpcds10.web_sales_bucketed[ws_item_sk#2] Batched:
> true, Format: Parquet, Location:...
> +- *(3) Sort [i_item_sk#6 ASC NULLS FIRST], false, 0
> +- Exchange hashpartitioning(i_item_sk#6, 4)
> +- *(2) Project [i_item_sk#6]
> +- *(2) Filter isnotnull(i_item_sk#6)
> +- *(2) FileScan parquet tpcds10.item[i_item_sk#6] Batched: true, Format:
> Parquet, Location:...{code}
> This problem could be worse if we enable the adaptive execution, because it
> usually prefers a big shuffle.parititon.
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
