[jira] [Commented] (SPARK-22276) Unnecessary repartitioning
[
https://issues.apache.org/jira/browse/SPARK-22276?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16206745#comment-16206745
]
Liang-Chi Hsieh commented on SPARK-22276:
-
I think this issue is already resolved by a recent fix at SPARK-3.
{code}
>>> from pyspark.sql import functions as F
>>> df =
>>> spark.range(1000).toDF("nr").select((F.rand()*100).cast("int").alias("pre_gid"),
>>> (F.rand()*100).cast("int").alias("post_gid"))
>>> df.sort("post_gid").groupBy("post_gid").agg(F.collect_list("pre_gid").alias("pre_gids")).explain()
== Physical Plan ==
ObjectHashAggregate(keys=[post_gid#29], functions=[collect_list(pre_gid#28, 0,
0)])
+- ObjectHashAggregate(keys=[post_gid#29],
functions=[partial_collect_list(pre_gid#28, 0, 0)])
+- *Sort [post_gid#29 ASC NULLS FIRST], true, 0
+- Exchange rangepartitioning(post_gid#29 ASC NULLS FIRST, 200)
+- *Project [cast((rand(-2985866240213757903) * 100.0) as int) AS
pre_gid#28, cast((rand(4357680211635473806) * 100.0) as int) AS post_gid#29]
+- *Range (0, 1000, step=1, splits=2)
{code}
> Unnecessary repartitioning
> --
>
> Key: SPARK-22276
> URL: https://issues.apache.org/jira/browse/SPARK-22276
> Project: Spark
> Issue Type: Bug
> Components: Optimizer
>Affects Versions: 2.2.0
>Reporter: Fernando Pereira
>
> When a dataframe is sorted it is partitioned with a RangePartitioner.
> If later we aggregate by the exact same fields over which sort was applied
> there is a new (apparently useless) Exchange repartitioning by a
> HashPartitioner.
> In my use case the groupBy exchange is still very costly as the aggregate
> function won't reduce the data volume.
> Is there any reason why groupBy always shuffles data, or could this be
> improved?
> Is there currently a way to workaround for the moment, without going to
> mapPartitions?
> Example
> {code}
> nrn_vals.printSchema()
> (nrn_vals
> .sort("post_gid")
> .groupBy("post_gid")
> .agg(F.collect_list("pre_gid").alias("pre_gids"))
> ).explain()
> {code}
> Outputs the following
> {code}
> root
> |-- pre_gid: integer (nullable = true)
> |-- post_gid: integer (nullable = true)
> |-- floatvec: array (nullable = false)
> ||-- element: float (containsNull = true)
> == Physical Plan ==
> ObjectHashAggregate(keys=[post_gid#1386],
> functions=[collect_list(pre_gid#1385, 0, 0)])
> +- Exchange hashpartitioning(post_gid#1386, 1)
>+- ObjectHashAggregate(keys=[post_gid#1386],
> functions=[partial_collect_list(pre_gid#1385, 0, 0)])
> +- *Sort [post_gid#1386 ASC NULLS FIRST], true, 0
> +- Exchange rangepartitioning(post_gid#1386 ASC NULLS FIRST, 1)
> +- *FileScan parquet [pre_gid#1385,post_gid#1386] Batched: true,
> Format: Parquet, Location:
> InMemoryFileIndex[file:/media/psf/Home/dev/Functionalizer/pyspark/spykfunc_output/extended_touche...,
> PartitionFilters: [], PushedFilters: [], ReadSchema:
> struct
> {code}
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
-
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-22276) Unnecessary repartitioning
[
https://issues.apache.org/jira/browse/SPARK-22276?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16205794#comment-16205794
]
Fernando Pereira commented on SPARK-22276:
--
I made some more tests and this behavior only appeared in Spark 2.2.0. In Spark
2.1.1 it works as I would expect.
> Unnecessary repartitioning
> --
>
> Key: SPARK-22276
> URL: https://issues.apache.org/jira/browse/SPARK-22276
> Project: Spark
> Issue Type: Bug
> Components: Optimizer
>Affects Versions: 2.2.0
>Reporter: Fernando Pereira
>
> When a dataframe is sorted it is partitioned with a RangePartitioner.
> If later we aggregate by the exact same fields over which sort was applied
> there is a new (apparently useless) Exchange repartitioning by a
> HashPartitioner.
> In my use case the groupBy exchange is still very costly as the aggregate
> function won't reduce the data volume.
> Is there any reason why groupBy always shuffles data, or could this be
> improved?
> Is there currently a way to workaround for the moment, without going to
> mapPartitions?
> Example
> {code}
> nrn_vals.printSchema()
> (nrn_vals
> .sort("post_gid")
> .groupBy("post_gid")
> .agg(F.collect_list("pre_gid").alias("pre_gids"))
> ).explain()
> {code}
> Outputs the following
> {code}
> root
> |-- pre_gid: integer (nullable = true)
> |-- post_gid: integer (nullable = true)
> |-- floatvec: array (nullable = false)
> ||-- element: float (containsNull = true)
> == Physical Plan ==
> ObjectHashAggregate(keys=[post_gid#1386],
> functions=[collect_list(pre_gid#1385, 0, 0)])
> +- Exchange hashpartitioning(post_gid#1386, 1)
>+- ObjectHashAggregate(keys=[post_gid#1386],
> functions=[partial_collect_list(pre_gid#1385, 0, 0)])
> +- *Sort [post_gid#1386 ASC NULLS FIRST], true, 0
> +- Exchange rangepartitioning(post_gid#1386 ASC NULLS FIRST, 1)
> +- *FileScan parquet [pre_gid#1385,post_gid#1386] Batched: true,
> Format: Parquet, Location:
> InMemoryFileIndex[file:/media/psf/Home/dev/Functionalizer/pyspark/spykfunc_output/extended_touche...,
> PartitionFilters: [], PushedFilters: [], ReadSchema:
> struct
> {code}
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
-
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-22276) Unnecessary repartitioning
[
https://issues.apache.org/jira/browse/SPARK-22276?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16205655#comment-16205655
]
Fernando Pereira commented on SPARK-22276:
--
I added a simple example that shows what I mean.
You see two exchanges, even though data is already partitioned according to the
same field. I wonder if the Exchange hashpartitioning is really needed, since
in my case it just slows things down.
Thanks
> Unnecessary repartitioning
> --
>
> Key: SPARK-22276
> URL: https://issues.apache.org/jira/browse/SPARK-22276
> Project: Spark
> Issue Type: Bug
> Components: Optimizer
>Affects Versions: 2.2.0
>Reporter: Fernando Pereira
>
> When a dataframe is sorted it is partitioned with a RangePartitioner.
> If later we aggregate by the exact same fields over which sort was applied
> there is a new (apparently useless) Exchange repartitioning by a
> HashPartitioner.
> In my use case the groupBy exchange is still very costly as the aggregate
> function won't reduce the data volume.
> Is there any reason why groupBy always shuffles data, or could this be
> improved?
> Is there currently a way to workaround for the moment, without going to
> mapPartitions?
> Example
> {code}
> nrn_vals.printSchema()
> (nrn_vals
> .sort("post_gid")
> .groupBy("post_gid")
> .agg(F.collect_list("pre_gid").alias("pre_gids"))
> ).explain()
> {code}
> Outputs the following
> {code}
> root
> |-- pre_gid: integer (nullable = true)
> |-- post_gid: integer (nullable = true)
> |-- floatvec: array (nullable = false)
> ||-- element: float (containsNull = true)
> == Physical Plan ==
> ObjectHashAggregate(keys=[post_gid#1386],
> functions=[collect_list(pre_gid#1385, 0, 0)])
> +- Exchange hashpartitioning(post_gid#1386, 1)
>+- ObjectHashAggregate(keys=[post_gid#1386],
> functions=[partial_collect_list(pre_gid#1385, 0, 0)])
> +- *Sort [post_gid#1386 ASC NULLS FIRST], true, 0
> +- Exchange rangepartitioning(post_gid#1386 ASC NULLS FIRST, 1)
> +- *FileScan parquet [pre_gid#1385,post_gid#1386] Batched: true,
> Format: Parquet, Location:
> InMemoryFileIndex[file:/media/psf/Home/dev/Functionalizer/pyspark/spykfunc_output/extended_touche...,
> PartitionFilters: [], PushedFilters: [], ReadSchema:
> struct
> {code}
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
-
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-22276) Unnecessary repartitioning
[ https://issues.apache.org/jira/browse/SPARK-22276?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16205400#comment-16205400 ] Liang-Chi Hsieh commented on SPARK-22276: - Can you provide an simple example to reproduce this issue? Then it is easier to verify the problem. > Unnecessary repartitioning > -- > > Key: SPARK-22276 > URL: https://issues.apache.org/jira/browse/SPARK-22276 > Project: Spark > Issue Type: Bug > Components: Optimizer >Affects Versions: 2.2.0 >Reporter: Fernando Pereira > > When a dataframe is sorted it is partitioned with a RangePartitioner. > If later we aggregate by the exact same fields over which sort was applied > there is a new (apparently useless) Exchange repartitioning by a > HashPartitioner. > In my use case the groupBy exchange is still very costly as the aggregate > function won't reduce the data volume. > Is there any reason why groupBy always shuffles data, or could this be > improved? > Is there currently a way to workaround for the moment, without going to > mapPartitions? -- This message was sent by Atlassian JIRA (v6.4.14#64029) - To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org
