[
https://issues.apache.org/jira/browse/SPARK-20169?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16418282#comment-16418282
]
Maryann Xue commented on SPARK-20169:
-------------------------------------
This was related to the use of "withColumnRenamed", which will add a "Project"
on top of the original relation. And later on, when getting the
"outputPartitioning" of the "Project", it returned HashPartitioning("dst")
while it should return HashPartitioning("src") since it had been renamed. As a
result, an "Exchange" node was missing between the two outmost HashAggregate
nodes, and that's why the aggregate result was not correct.
{code:java}
test("SPARK-20169") {
val e = Seq((1, 2), (1, 3), (1, 4), (2, 1), (3, 1), (4, 1)).toDF("src", "dst")
val r = Seq((1), (2), (3), (4)).toDF("src")
val r1 = e.join(r, "src" ::
Nil).groupBy("dst").count().withColumnRenamed("dst", "src")
val jr = e.join(r1, "src" :: Nil)
val r2 = jr.groupBy("dst").count
r2.explain()
r2.show
}
{code}
The physical plan resulted from this bug turned out to be
{code}
== Physical Plan ==
*(2) HashAggregate(keys=[dst#181], functions=[count(1)])
+- *(2) HashAggregate(keys=[dst#181], functions=[partial_count(1)])
+- *(2) Project [dst#181]
+- *(2) BroadcastHashJoin [src#180], [src#197], Inner, BuildLeft
:- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0,
int, false] as bigint)))
: +- LocalTableScan [src#180, dst#181]
+- *(2) HashAggregate(keys=[dst#181], functions=[])
+- Exchange hashpartitioning(dst#181, 5)
+- *(1) HashAggregate(keys=[dst#181], functions=[])
+- *(1) Project [dst#181]
+- *(1) BroadcastHashJoin [src#180], [src#187], Inner,
BuildRight
:- LocalTableScan [src#180, dst#181]
+- BroadcastExchange
HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)))
+- LocalTableScan [src#187]
{code}
The correct physical plan should be
{code}
== Physical Plan ==
*(3) HashAggregate(keys=[dst#181], functions=[count(1)])
+- Exchange hashpartitioning(dst#181, 5)
+- *(2) HashAggregate(keys=[dst#181], functions=[partial_count(1)])
+- *(2) Project [dst#181]
+- *(2) BroadcastHashJoin [src#180], [src#197], Inner, BuildLeft
:- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0,
int, false] as bigint)))
: +- LocalTableScan [src#180, dst#181]
+- *(2) HashAggregate(keys=[dst#181], functions=[])
+- Exchange hashpartitioning(dst#181, 5)
+- *(1) HashAggregate(keys=[dst#181], functions=[])
+- *(1) Project [dst#181]
+- *(1) BroadcastHashJoin [src#180], [src#187], Inner,
BuildRight
:- LocalTableScan [src#180, dst#181]
+- BroadcastExchange
HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)))
+- LocalTableScan [src#187]
{code}
This happens to the same issue as SPARK-23368. The PR for SPARK-23368 fixes it.
> Groupby Bug with Sparksql
> -------------------------
>
> Key: SPARK-20169
> URL: https://issues.apache.org/jira/browse/SPARK-20169
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 2.0.0, 2.1.0
> Reporter: Bin Wu
> Priority: Major
>
> We find a potential bug in Catalyst optimizer which cannot correctly
> process "groupby". You can reproduce it by following simple example:
> =========================
> from pyspark.sql.functions import *
> #e=sc.parallelize([(1,2),(1,3),(1,4),(2,1),(3,1),(4,1)]).toDF(["src","dst"])
> e = spark.read.csv("graph.csv", header=True)
> r = sc.parallelize([(1,),(2,),(3,),(4,)]).toDF(['src'])
> r1 = e.join(r, 'src').groupBy('dst').count().withColumnRenamed('dst','src')
> jr = e.join(r1, 'src')
> jr.show()
> r2 = jr.groupBy('dst').count()
> r2.show()
> =========================
> FYI, "graph.csv" contains exactly the same data as the commented line.
> You can find that jr is:
> |src|dst|count|
> | 3| 1| 1|
> | 1| 4| 3|
> | 1| 3| 3|
> | 1| 2| 3|
> | 4| 1| 1|
> | 2| 1| 1|
> But, after the last groupBy, the 3 rows with dst = 1 are not grouped together:
> |dst|count|
> | 1| 1|
> | 4| 1|
> | 3| 1|
> | 2| 1|
> | 1| 1|
> | 1| 1|
> If we build jr directly from raw data (commented line), this error will not
> show up. So
> we suspect that there is a bug in the Catalyst optimizer when multiple joins
> and groupBy's
> are being optimized.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
