[jira] [Commented] (SPARK-18676) Spark 2.x query plan data size estimation can crash join queries versus 1.x
[
https://issues.apache.org/jira/browse/SPARK-18676?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15746414#comment-15746414
]
Reynold Xin commented on SPARK-18676:
-
That's the other option I was considering. It'd be good to decouple the
mechanism for data distribution (broadcast vs shuffle) and join method (hash vs
sort-merge). And both hash and sort-merge should work with data larger than
memory.
> Spark 2.x query plan data size estimation can crash join queries versus 1.x
> ---
>
> Key: SPARK-18676
> URL: https://issues.apache.org/jira/browse/SPARK-18676
> Project: Spark
> Issue Type: Bug
> Components: SQL
>Affects Versions: 2.0.0, 2.0.1, 2.0.2, 2.1.0
>Reporter: Michael Allman
>
> Commit [c481bdf|https://github.com/apache/spark/commit/c481bdf] significantly
> modified the way Spark SQL estimates the output data size of query plans.
> I've found that—with the new table query partition pruning support in
> 2.1—this has lead to in some cases underestimation of join plan child size
> statistics to a degree that makes executing such queries impossible without
> disabling automatic broadcast conversion.
> In one case we debugged, the query planner had estimated the size of a join
> child to be 3,854 bytes. In the execution of this child query, Spark reads 20
> million rows in 1 GB of data from parquet files and shuffles 722.9 MB of
> data, outputting 17 million rows. In planning the original join query, Spark
> converts the child to a {{BroadcastExchange}}. This query execution fails
> unless automatic broadcast conversion is disabled.
> This particular query is complex and very specific to our data and schema. I
> have not yet developed a reproducible test case that can be shared. I realize
> this ticket does not give the Spark team a lot to work with to reproduce and
> test this issue, but I'm available to help. At the moment I can suggest
> running a join where one side is an aggregation selecting a few fields over a
> large table with a wide schema including many string columns.
> This issue exists in Spark 2.0, but we never encountered it because in that
> version it only manifests itself for partitioned relations read from the
> filesystem, and we rarely use this feature. We've encountered this issue in
> 2.1 because 2.1 does partition pruning for metastore tables now.
> As a back stop, we've patched our branch of Spark 2.1 to revert the
> reductions in default data type size for string, binary and user-defined
> types. We also removed the override of the statistics method in {{UnaryNode}}
> which reduces the output size of a plan based on the ratio of that plan's
> output schema size versus its children's. We have not had this problem since.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
-
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-18676) Spark 2.x query plan data size estimation can crash join queries versus 1.x
[
https://issues.apache.org/jira/browse/SPARK-18676?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15746382#comment-15746382
]
Michael Allman commented on SPARK-18676:
Yeah, I was wondering how that would work with the UI/query plan. What if we
incorporated a disk-based fallback into the broadcast plans themselves?
> Spark 2.x query plan data size estimation can crash join queries versus 1.x
> ---
>
> Key: SPARK-18676
> URL: https://issues.apache.org/jira/browse/SPARK-18676
> Project: Spark
> Issue Type: Bug
> Components: SQL
>Affects Versions: 2.0.0, 2.0.1, 2.0.2, 2.1.0
>Reporter: Michael Allman
>
> Commit [c481bdf|https://github.com/apache/spark/commit/c481bdf] significantly
> modified the way Spark SQL estimates the output data size of query plans.
> I've found that—with the new table query partition pruning support in
> 2.1—this has lead to in some cases underestimation of join plan child size
> statistics to a degree that makes executing such queries impossible without
> disabling automatic broadcast conversion.
> In one case we debugged, the query planner had estimated the size of a join
> child to be 3,854 bytes. In the execution of this child query, Spark reads 20
> million rows in 1 GB of data from parquet files and shuffles 722.9 MB of
> data, outputting 17 million rows. In planning the original join query, Spark
> converts the child to a {{BroadcastExchange}}. This query execution fails
> unless automatic broadcast conversion is disabled.
> This particular query is complex and very specific to our data and schema. I
> have not yet developed a reproducible test case that can be shared. I realize
> this ticket does not give the Spark team a lot to work with to reproduce and
> test this issue, but I'm available to help. At the moment I can suggest
> running a join where one side is an aggregation selecting a few fields over a
> large table with a wide schema including many string columns.
> This issue exists in Spark 2.0, but we never encountered it because in that
> version it only manifests itself for partitioned relations read from the
> filesystem, and we rarely use this feature. We've encountered this issue in
> 2.1 because 2.1 does partition pruning for metastore tables now.
> As a back stop, we've patched our branch of Spark 2.1 to revert the
> reductions in default data type size for string, binary and user-defined
> types. We also removed the override of the statistics method in {{UnaryNode}}
> which reduces the output size of a plan based on the ratio of that plan's
> output schema size versus its children's. We have not had this problem since.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
-
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-18676) Spark 2.x query plan data size estimation can crash join queries versus 1.x
[
https://issues.apache.org/jira/browse/SPARK-18676?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15746370#comment-15746370
]
Davies Liu commented on SPARK-18676:
I had a working prototype, but in introduce some weird behavior, for example,
the actual plan will not match the one showed in explain or web ui.
Currently, I'm still not sure that the right direction or not.
> Spark 2.x query plan data size estimation can crash join queries versus 1.x
> ---
>
> Key: SPARK-18676
> URL: https://issues.apache.org/jira/browse/SPARK-18676
> Project: Spark
> Issue Type: Bug
> Components: SQL
>Affects Versions: 2.0.0, 2.0.1, 2.0.2, 2.1.0
>Reporter: Michael Allman
>
> Commit [c481bdf|https://github.com/apache/spark/commit/c481bdf] significantly
> modified the way Spark SQL estimates the output data size of query plans.
> I've found that—with the new table query partition pruning support in
> 2.1—this has lead to in some cases underestimation of join plan child size
> statistics to a degree that makes executing such queries impossible without
> disabling automatic broadcast conversion.
> In one case we debugged, the query planner had estimated the size of a join
> child to be 3,854 bytes. In the execution of this child query, Spark reads 20
> million rows in 1 GB of data from parquet files and shuffles 722.9 MB of
> data, outputting 17 million rows. In planning the original join query, Spark
> converts the child to a {{BroadcastExchange}}. This query execution fails
> unless automatic broadcast conversion is disabled.
> This particular query is complex and very specific to our data and schema. I
> have not yet developed a reproducible test case that can be shared. I realize
> this ticket does not give the Spark team a lot to work with to reproduce and
> test this issue, but I'm available to help. At the moment I can suggest
> running a join where one side is an aggregation selecting a few fields over a
> large table with a wide schema including many string columns.
> This issue exists in Spark 2.0, but we never encountered it because in that
> version it only manifests itself for partitioned relations read from the
> filesystem, and we rarely use this feature. We've encountered this issue in
> 2.1 because 2.1 does partition pruning for metastore tables now.
> As a back stop, we've patched our branch of Spark 2.1 to revert the
> reductions in default data type size for string, binary and user-defined
> types. We also removed the override of the statistics method in {{UnaryNode}}
> which reduces the output size of a plan based on the ratio of that plan's
> output schema size versus its children's. We have not had this problem since.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
-
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-18676) Spark 2.x query plan data size estimation can crash join queries versus 1.x
[
https://issues.apache.org/jira/browse/SPARK-18676?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15743362#comment-15743362
]
Reynold Xin commented on SPARK-18676:
-
Can we just increase the size by 5X if it is a Parquet or ORC file?
> Spark 2.x query plan data size estimation can crash join queries versus 1.x
> ---
>
> Key: SPARK-18676
> URL: https://issues.apache.org/jira/browse/SPARK-18676
> Project: Spark
> Issue Type: Bug
> Components: SQL
>Affects Versions: 2.0.0, 2.0.1, 2.0.2, 2.1.0
>Reporter: Michael Allman
>
> Commit [c481bdf|https://github.com/apache/spark/commit/c481bdf] significantly
> modified the way Spark SQL estimates the output data size of query plans.
> I've found that—with the new table query partition pruning support in
> 2.1—this has lead to in some cases underestimation of join plan child size
> statistics to a degree that makes executing such queries impossible without
> disabling automatic broadcast conversion.
> In one case we debugged, the query planner had estimated the size of a join
> child to be 3,854 bytes. In the execution of this child query, Spark reads 20
> million rows in 1 GB of data from parquet files and shuffles 722.9 MB of
> data, outputting 17 million rows. In planning the original join query, Spark
> converts the child to a {{BroadcastExchange}}. This query execution fails
> unless automatic broadcast conversion is disabled.
> This particular query is complex and very specific to our data and schema. I
> have not yet developed a reproducible test case that can be shared. I realize
> this ticket does not give the Spark team a lot to work with to reproduce and
> test this issue, but I'm available to help. At the moment I can suggest
> running a join where one side is an aggregation selecting a few fields over a
> large table with a wide schema including many string columns.
> This issue exists in Spark 2.0, but we never encountered it because in that
> version it only manifests itself for partitioned relations read from the
> filesystem, and we rarely use this feature. We've encountered this issue in
> 2.1 because 2.1 does partition pruning for metastore tables now.
> As a back stop, we've patched our branch of Spark 2.1 to revert the
> reductions in default data type size for string, binary and user-defined
> types. We also removed the override of the statistics method in {{UnaryNode}}
> which reduces the output size of a plan based on the ratio of that plan's
> output schema size versus its children's. We have not had this problem since.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
-
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-18676) Spark 2.x query plan data size estimation can crash join queries versus 1.x
[
https://issues.apache.org/jira/browse/SPARK-18676?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15743076#comment-15743076
]
Michael Allman commented on SPARK-18676:
I'm sorry I have not had time to provide more information on my findings. I can
summarize by saying that none of the source patch variations I tried provided
an accurate estimate. Making this estimate more accurate could be a good
project for 2.2.
In the absence of an accurate size estimate, [~davies]'s idea for switching to
ShuffleJoin for oversized broadcasts sounds like a good idea. [~davies], is
that something you'd like to work on?
> Spark 2.x query plan data size estimation can crash join queries versus 1.x
> ---
>
> Key: SPARK-18676
> URL: https://issues.apache.org/jira/browse/SPARK-18676
> Project: Spark
> Issue Type: Bug
> Components: SQL
>Affects Versions: 2.0.0, 2.0.1, 2.0.2, 2.1.0
>Reporter: Michael Allman
>
> Commit [c481bdf|https://github.com/apache/spark/commit/c481bdf] significantly
> modified the way Spark SQL estimates the output data size of query plans.
> I've found that—with the new table query partition pruning support in
> 2.1—this has lead to in some cases underestimation of join plan child size
> statistics to a degree that makes executing such queries impossible without
> disabling automatic broadcast conversion.
> In one case we debugged, the query planner had estimated the size of a join
> child to be 3,854 bytes. In the execution of this child query, Spark reads 20
> million rows in 1 GB of data from parquet files and shuffles 722.9 MB of
> data, outputting 17 million rows. In planning the original join query, Spark
> converts the child to a {{BroadcastExchange}}. This query execution fails
> unless automatic broadcast conversion is disabled.
> This particular query is complex and very specific to our data and schema. I
> have not yet developed a reproducible test case that can be shared. I realize
> this ticket does not give the Spark team a lot to work with to reproduce and
> test this issue, but I'm available to help. At the moment I can suggest
> running a join where one side is an aggregation selecting a few fields over a
> large table with a wide schema including many string columns.
> This issue exists in Spark 2.0, but we never encountered it because in that
> version it only manifests itself for partitioned relations read from the
> filesystem, and we rarely use this feature. We've encountered this issue in
> 2.1 because 2.1 does partition pruning for metastore tables now.
> As a back stop, we've patched our branch of Spark 2.1 to revert the
> reductions in default data type size for string, binary and user-defined
> types. We also removed the override of the statistics method in {{UnaryNode}}
> which reduces the output size of a plan based on the ratio of that plan's
> output schema size versus its children's. We have not had this problem since.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
-
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-18676) Spark 2.x query plan data size estimation can crash join queries versus 1.x
[
https://issues.apache.org/jira/browse/SPARK-18676?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15734560#comment-15734560
]
Michael Allman commented on SPARK-18676:
Ah okay. That might be a strategy to explore.
> Spark 2.x query plan data size estimation can crash join queries versus 1.x
> ---
>
> Key: SPARK-18676
> URL: https://issues.apache.org/jira/browse/SPARK-18676
> Project: Spark
> Issue Type: Bug
> Components: SQL
>Affects Versions: 2.0.0, 2.0.1, 2.0.2, 2.1.0
>Reporter: Michael Allman
>
> Commit [c481bdf|https://github.com/apache/spark/commit/c481bdf] significantly
> modified the way Spark SQL estimates the output data size of query plans.
> I've found that—with the new table query partition pruning support in
> 2.1—this has lead to in some cases underestimation of join plan child size
> statistics to a degree that makes executing such queries impossible without
> disabling automatic broadcast conversion.
> In one case we debugged, the query planner had estimated the size of a join
> child to be 3,854 bytes. In the execution of this child query, Spark reads 20
> million rows in 1 GB of data from parquet files and shuffles 722.9 MB of
> data, outputting 17 million rows. In planning the original join query, Spark
> converts the child to a {{BroadcastExchange}}. This query execution fails
> unless automatic broadcast conversion is disabled.
> This particular query is complex and very specific to our data and schema. I
> have not yet developed a reproducible test case that can be shared. I realize
> this ticket does not give the Spark team a lot to work with to reproduce and
> test this issue, but I'm available to help. At the moment I can suggest
> running a join where one side is an aggregation selecting a few fields over a
> large table with a wide schema including many string columns.
> This issue exists in Spark 2.0, but we never encountered it because in that
> version it only manifests itself for partitioned relations read from the
> filesystem, and we rarely use this feature. We've encountered this issue in
> 2.1 because 2.1 does partition pruning for metastore tables now.
> As a back stop, we've patched our branch of Spark 2.1 to revert the
> reductions in default data type size for string, binary and user-defined
> types. We also removed the override of the statistics method in {{UnaryNode}}
> which reduces the output size of a plan based on the ratio of that plan's
> output schema size versus its children's. We have not had this problem since.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
-
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-18676) Spark 2.x query plan data size estimation can crash join queries versus 1.x
[
https://issues.apache.org/jira/browse/SPARK-18676?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15733569#comment-15733569
]
Davies Liu commented on SPARK-18676:
Yes, it can, see WholeStageCodegen.doExecute() as an example.
Phisical plan is used by generate RDD (which is the actual physical plan), we
build another SparkPlan to generate the RDD in the middle.
> Spark 2.x query plan data size estimation can crash join queries versus 1.x
> ---
>
> Key: SPARK-18676
> URL: https://issues.apache.org/jira/browse/SPARK-18676
> Project: Spark
> Issue Type: Bug
> Components: SQL
>Affects Versions: 2.0.0, 2.0.1, 2.0.2, 2.1.0
>Reporter: Michael Allman
>
> Commit [c481bdf|https://github.com/apache/spark/commit/c481bdf] significantly
> modified the way Spark SQL estimates the output data size of query plans.
> I've found that—with the new table query partition pruning support in
> 2.1—this has lead to in some cases underestimation of join plan child size
> statistics to a degree that makes executing such queries impossible without
> disabling automatic broadcast conversion.
> In one case we debugged, the query planner had estimated the size of a join
> child to be 3,854 bytes. In the execution of this child query, Spark reads 20
> million rows in 1 GB of data from parquet files and shuffles 722.9 MB of
> data, outputting 17 million rows. In planning the original join query, Spark
> converts the child to a {{BroadcastExchange}}. This query execution fails
> unless automatic broadcast conversion is disabled.
> This particular query is complex and very specific to our data and schema. I
> have not yet developed a reproducible test case that can be shared. I realize
> this ticket does not give the Spark team a lot to work with to reproduce and
> test this issue, but I'm available to help. At the moment I can suggest
> running a join where one side is an aggregation selecting a few fields over a
> large table with a wide schema including many string columns.
> This issue exists in Spark 2.0, but we never encountered it because in that
> version it only manifests itself for partitioned relations read from the
> filesystem, and we rarely use this feature. We've encountered this issue in
> 2.1 because 2.1 does partition pruning for metastore tables now.
> As a back stop, we've patched our branch of Spark 2.1 to revert the
> reductions in default data type size for string, binary and user-defined
> types. We also removed the override of the statistics method in {{UnaryNode}}
> which reduces the output size of a plan based on the ratio of that plan's
> output schema size versus its children's. We have not had this problem since.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
-
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-18676) Spark 2.x query plan data size estimation can crash join queries versus 1.x
[
https://issues.apache.org/jira/browse/SPARK-18676?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15726437#comment-15726437
]
Michael Allman commented on SPARK-18676:
> maybe we could switch to ShuffleJoin when it realize that the actual data is
> larger than thought
I'm not sure what you mean here. Can the query execution change its join
strategy in the middle of executing it?
> Spark 2.x query plan data size estimation can crash join queries versus 1.x
> ---
>
> Key: SPARK-18676
> URL: https://issues.apache.org/jira/browse/SPARK-18676
> Project: Spark
> Issue Type: Bug
> Components: SQL
>Affects Versions: 2.0.0, 2.0.1, 2.0.2, 2.1.0
>Reporter: Michael Allman
>
> Commit [c481bdf|https://github.com/apache/spark/commit/c481bdf] significantly
> modified the way Spark SQL estimates the output data size of query plans.
> I've found that—with the new table query partition pruning support in
> 2.1—this has lead to in some cases underestimation of join plan child size
> statistics to a degree that makes executing such queries impossible without
> disabling automatic broadcast conversion.
> In one case we debugged, the query planner had estimated the size of a join
> child to be 3,854 bytes. In the execution of this child query, Spark reads 20
> million rows in 1 GB of data from parquet files and shuffles 722.9 MB of
> data, outputting 17 million rows. In planning the original join query, Spark
> converts the child to a {{BroadcastExchange}}. This query execution fails
> unless automatic broadcast conversion is disabled.
> This particular query is complex and very specific to our data and schema. I
> have not yet developed a reproducible test case that can be shared. I realize
> this ticket does not give the Spark team a lot to work with to reproduce and
> test this issue, but I'm available to help. At the moment I can suggest
> running a join where one side is an aggregation selecting a few fields over a
> large table with a wide schema including many string columns.
> This issue exists in Spark 2.0, but we never encountered it because in that
> version it only manifests itself for partitioned relations read from the
> filesystem, and we rarely use this feature. We've encountered this issue in
> 2.1 because 2.1 does partition pruning for metastore tables now.
> As a back stop, we've patched our branch of Spark 2.1 to revert the
> reductions in default data type size for string, binary and user-defined
> types. We also removed the override of the statistics method in {{UnaryNode}}
> which reduces the output size of a plan based on the ratio of that plan's
> output schema size versus its children's. We have not had this problem since.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
-
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-18676) Spark 2.x query plan data size estimation can crash join queries versus 1.x
[
https://issues.apache.org/jira/browse/SPARK-18676?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15726431#comment-15726431
]
Michael Allman commented on SPARK-18676:
I'm spending some more time this week to understand where things are going
wrong in our query size estimate. I'll have a more detailed analysis to share
later today or tomorrow. As you mentioned, parquet file size is not a good
estimate for data size. If I have time, I'll compare the Spark data size to the
parquet file column statistics.
Another thing I can say right now is that the array type default size estimate
is very inaccurate for some of our biggest tables. Spark multiples the array
type element's data size estimate by 100 to get the array type's estimate. In
our case, we have a struct type with an array field with a struct type with an
array field. In practice, these arrays never have more than one element in
them. So that's leading to a four order of magnitude error in Spark's size
estimate of those tables' types.
Also, for the purposes of estimating data size, I believe it's incorrect to
multiply array and map type estimates by 100. The underlying table's estimate
already factors in the size of those columns, including all of their array and
map elements. The next thing I'm testing is eliminating those factors of 100 in
the size estimates.
> Spark 2.x query plan data size estimation can crash join queries versus 1.x
> ---
>
> Key: SPARK-18676
> URL: https://issues.apache.org/jira/browse/SPARK-18676
> Project: Spark
> Issue Type: Bug
> Components: SQL
>Affects Versions: 2.0.0, 2.0.1, 2.0.2, 2.1.0
>Reporter: Michael Allman
>
> Commit [c481bdf|https://github.com/apache/spark/commit/c481bdf] significantly
> modified the way Spark SQL estimates the output data size of query plans.
> I've found that—with the new table query partition pruning support in
> 2.1—this has lead to in some cases underestimation of join plan child size
> statistics to a degree that makes executing such queries impossible without
> disabling automatic broadcast conversion.
> In one case we debugged, the query planner had estimated the size of a join
> child to be 3,854 bytes. In the execution of this child query, Spark reads 20
> million rows in 1 GB of data from parquet files and shuffles 722.9 MB of
> data, outputting 17 million rows. In planning the original join query, Spark
> converts the child to a {{BroadcastExchange}}. This query execution fails
> unless automatic broadcast conversion is disabled.
> This particular query is complex and very specific to our data and schema. I
> have not yet developed a reproducible test case that can be shared. I realize
> this ticket does not give the Spark team a lot to work with to reproduce and
> test this issue, but I'm available to help. At the moment I can suggest
> running a join where one side is an aggregation selecting a few fields over a
> large table with a wide schema including many string columns.
> This issue exists in Spark 2.0, but we never encountered it because in that
> version it only manifests itself for partitioned relations read from the
> filesystem, and we rarely use this feature. We've encountered this issue in
> 2.1 because 2.1 does partition pruning for metastore tables now.
> As a back stop, we've patched our branch of Spark 2.1 to revert the
> reductions in default data type size for string, binary and user-defined
> types. We also removed the override of the statistics method in {{UnaryNode}}
> which reduces the output size of a plan based on the ratio of that plan's
> output schema size versus its children's. We have not had this problem since.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
-
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-18676) Spark 2.x query plan data size estimation can crash join queries versus 1.x
[
https://issues.apache.org/jira/browse/SPARK-18676?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15726247#comment-15726247
]
Davies Liu commented on SPARK-18676:
What's the schema and plan of the child looks like? It's possible that the
schema of parquet file is wide, also highly compressed, only a few column are
used in the query, then the estimation will be much smaller than actual data
size. The estimation of string could also be wrong.
Using the bytes of parquet file as the metric for broadcasting is bad, we also
saw some cases that the parquet file is only a few MB, but the broadcast is a
few GB.
The estimation could easily be wrong for many reasons, maybe we could switch to
ShuffleJoin when it realize that the actual data is larger than thought, will
that work?
> Spark 2.x query plan data size estimation can crash join queries versus 1.x
> ---
>
> Key: SPARK-18676
> URL: https://issues.apache.org/jira/browse/SPARK-18676
> Project: Spark
> Issue Type: Bug
> Components: SQL
>Affects Versions: 2.0.0, 2.0.1, 2.0.2, 2.1.0
>Reporter: Michael Allman
>
> Commit [c481bdf|https://github.com/apache/spark/commit/c481bdf] significantly
> modified the way Spark SQL estimates the output data size of query plans.
> I've found that—with the new table query partition pruning support in
> 2.1—this has lead to in some cases underestimation of join plan child size
> statistics to a degree that makes executing such queries impossible without
> disabling automatic broadcast conversion.
> In one case we debugged, the query planner had estimated the size of a join
> child to be 3,854 bytes. In the execution of this child query, Spark reads 20
> million rows in 1 GB of data from parquet files and shuffles 722.9 MB of
> data, outputting 17 million rows. In planning the original join query, Spark
> converts the child to a {{BroadcastExchange}}. This query execution fails
> unless automatic broadcast conversion is disabled.
> This particular query is complex and very specific to our data and schema. I
> have not yet developed a reproducible test case that can be shared. I realize
> this ticket does not give the Spark team a lot to work with to reproduce and
> test this issue, but I'm available to help. At the moment I can suggest
> running a join where one side is an aggregation selecting a few fields over a
> large table with a wide schema including many string columns.
> This issue exists in Spark 2.0, but we never encountered it because in that
> version it only manifests itself for partitioned relations read from the
> filesystem, and we rarely use this feature. We've encountered this issue in
> 2.1 because 2.1 does partition pruning for metastore tables now.
> As a back stop, we've patched our branch of Spark 2.1 to revert the
> reductions in default data type size for string, binary and user-defined
> types. We also removed the override of the statistics method in `LogicalPlan`
> which reduces the output size of a plan based on the ratio of that plan's
> output schema size versus its children's. We have not had this problem since.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
-
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-18676) Spark 2.x query plan data size estimation can crash join queries versus 1.x
[
https://issues.apache.org/jira/browse/SPARK-18676?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15712823#comment-15712823
]
Michael Allman commented on SPARK-18676:
cc [~davies] as author of https://github.com/apache/spark/commit/c481bdf
> Spark 2.x query plan data size estimation can crash join queries versus 1.x
> ---
>
> Key: SPARK-18676
> URL: https://issues.apache.org/jira/browse/SPARK-18676
> Project: Spark
> Issue Type: Bug
> Components: SQL
>Affects Versions: 2.0.0, 2.0.1, 2.0.2, 2.1.0
>Reporter: Michael Allman
>
> Commit [c481bdf|https://github.com/apache/spark/commit/c481bdf] significantly
> modified the way Spark SQL estimates the output data size of query plans.
> I've found that—with the new table query partition pruning support in
> 2.1—this has lead to in some cases underestimation of join plan child size
> statistics to a degree that makes executing such queries impossible without
> disabling automatic broadcast conversion.
> In one case we debugged, the query planner had estimated the size of a join
> child to be 3,854 bytes. In the execution of this child query, Spark reads 20
> million rows in 1 GB of data from parquet files and shuffles 722.9 MB of
> data, outputting 17 million rows. In planning the original join query, Spark
> converts the child to a {{BroadcastExchange}}. This query execution fails
> unless automatic broadcast conversion is disabled.
> This particular query is complex and very specific to our data and schema. I
> have not yet developed a reproducible test case that can be shared. I realize
> this ticket does not give the Spark team a lot to work with to reproduce and
> test this issue, but I'm available to help. At the moment I can suggest
> running a join where one side is an aggregation selecting a few fields over a
> large table with a wide schema including many string columns.
> This issue exists in Spark 2.0, but we never encountered it because in that
> version it only manifests itself for partitioned relations read from the
> filesystem, and we rarely use this feature. We've encountered this issue in
> 2.1 because 2.1 does partition pruning for metastore tables now.
> As a back stop, we've patched our branch of Spark 2.1 to revert the
> reductions in default data type size for string, binary and user-defined
> types. We also removed the override of the statistics method in `LogicalPlan`
> which reduces the output size of a plan based on the ratio of that plan's
> output schema size versus its children's.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
-
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
