You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "xi-db (via GitHub)" <gi...@apache.org> on 2024/04/11 17:51:09 UTC
[PR] [WIP][SPARK-47818] Introduce plan cache in SparkConnectPlanner to improve performance of Analyze requests [spark]
xi-db opened a new pull request, #46012:
URL: https://github.com/apache/spark/pull/46012
### What changes were proposed in this pull request?
While building the DataFrame step by step, each time a new DataFrame is generated with an empty schema, which is lazily computed on access. However, if a user's code frequently accesses the schema of these new DataFrames using methods such as `df.columns`, it will result in a large number of Analyze requests to the server. Each time, the entire plan needs to be reanalyzed, leading to poor performance, especially when constructing highly complex plans.
Now, by introducing plan cache in SparkConnectPlanner, we aim to reduce the overhead of repeated analysis during this process. This is achieved by saving significant computation if the resolved logical plan of a subtree of can be cached.
A minimal example of the problem:
```
import pyspark.sql.functions as F
df = spark.range(10)
for i in range(200):
if str(i) not in df.columns: # <-- The df.columns call causes a new Analyze request in every iteration
df = df.withColumn(str(i), F.col("id") + i)
df.show()
```
With this patch, the performance of the above code improved from ~115s to ~5s.
### Why are the changes needed?
The performance improvement is huge.
### Does this PR introduce _any_ user-facing change?
Yes, a static conf `spark.connect.session.planCache.maxSize` and a dynamic conf `spark.connect.session.planCache.enabled` are added.
* `spark.connect.session.planCache.maxSize`: Sets the maximum number of cached resolved logical plans in Spark Connect Session. If set to a value less or equal than zero will disable the plan cache
* `spark.connect.session.planCache.enabled`: When true, the cache of resolved logical plans is enabled if `spark.connect.session.planCache.maxSize` is greater than zero. When false, the cache is disabled even if `spark.connect.session.planCache.maxSize` is greater than zero. The caching is best-effort and not guaranteed.
### How was this patch tested?
Some new tests are added in SparkConnectSessionHolderSuite.scala.
### Was this patch authored or co-authored using generative AI tooling?
No.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-47818][CONNECT] Introduce plan cache in SparkConnectPlanner to improve performance of Analyze requests [spark]
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #46012:
URL: https://github.com/apache/spark/pull/46012#discussion_r1562547428
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SessionHolder.scala:
##########
@@ -381,6 +405,53 @@ case class SessionHolder(userId: String, sessionId: String, session: SparkSessio
*/
private[connect] val pythonAccumulator: Option[PythonAccumulator] =
Try(session.sparkContext.collectionAccumulator[Array[Byte]]).toOption
+
+ /**
+ * Transform a relation into a logical plan, using the plan cache if enabled.
+ * The plan cache is enable only if `spark.connect.session.planCache.maxSize` is greater than zero
+ * AND `spark.connect.session.planCache.enabled` is true.
+ * @param rel The relation to transform.
+ * @param cachePlan Whether to cache the result logical plan.
+ * @param transform Function to transform the relation into a logical plan.
+ * @return The logical plan.
+ */
+ private[connect] def usePlanCache(rel: proto.Relation, cachePlan: Boolean)(
Review Comment:
I think we may want to exclude some kinds of plans:
1, LOCAL_RELATION, its size might be large;
2, CACHED_REMOTE_RELATION, it is already cached in `dataFrameCache` in the `SessionHolder`;
3, CATALOG, it may be a CREATE_TABLE command, should not be skipped.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-47818][CONNECT] Introduce plan cache in SparkConnectPlanner to improve performance of Analyze requests [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon closed pull request #46012: [SPARK-47818][CONNECT] Introduce plan cache in SparkConnectPlanner to improve performance of Analyze requests
URL: https://github.com/apache/spark/pull/46012
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-47818][CONNECT] Introduce plan cache in SparkConnectPlanner to improve performance of Analyze requests [spark]
Posted by "xi-db (via GitHub)" <gi...@apache.org>.
xi-db commented on code in PR #46012:
URL: https://github.com/apache/spark/pull/46012#discussion_r1565806605
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SessionHolder.scala:
##########
@@ -381,6 +405,53 @@ case class SessionHolder(userId: String, sessionId: String, session: SparkSessio
*/
private[connect] val pythonAccumulator: Option[PythonAccumulator] =
Try(session.sparkContext.collectionAccumulator[Array[Byte]]).toOption
+
+ /**
+ * Transform a relation into a logical plan, using the plan cache if enabled.
+ * The plan cache is enable only if `spark.connect.session.planCache.maxSize` is greater than zero
+ * AND `spark.connect.session.planCache.enabled` is true.
+ * @param rel The relation to transform.
+ * @param cachePlan Whether to cache the result logical plan.
+ * @param transform Function to transform the relation into a logical plan.
+ * @return The logical plan.
+ */
+ private[connect] def usePlanCache(rel: proto.Relation, cachePlan: Boolean)(
Review Comment:
> 3, CATALOG, it may be a command like CREATE_TABLE/DROP_TEMP_VIEW , should not be skipped.
Commands should be fine. Each command will get a new `plan_id`, so a new command won't hit the cache. Besides, we only cache plans with `plan_id` set.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-47818][CONNECT] Introduce plan cache in SparkConnectPlanner to improve performance of Analyze requests [spark]
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on code in PR #46012:
URL: https://github.com/apache/spark/pull/46012#discussion_r1565497107
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SessionHolder.scala:
##########
@@ -381,6 +405,53 @@ case class SessionHolder(userId: String, sessionId: String, session: SparkSessio
*/
private[connect] val pythonAccumulator: Option[PythonAccumulator] =
Try(session.sparkContext.collectionAccumulator[Array[Byte]]).toOption
+
+ /**
+ * Transform a relation into a logical plan, using the plan cache if enabled.
+ * The plan cache is enable only if `spark.connect.session.planCache.maxSize` is greater than zero
+ * AND `spark.connect.session.planCache.enabled` is true.
+ * @param rel The relation to transform.
+ * @param cachePlan Whether to cache the result logical plan.
+ * @param transform Function to transform the relation into a logical plan.
+ * @return The logical plan.
+ */
+ private[connect] def usePlanCache(rel: proto.Relation, cachePlan: Boolean)(
Review Comment:
@zhengruifeng Does point 3 extend to all DDL/DML commands?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-47818][CONNECT] Introduce plan cache in SparkConnectPlanner to improve performance of Analyze requests [spark]
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on PR #46012:
URL: https://github.com/apache/spark/pull/46012#issuecomment-2051344202
cc @HyukjinKwon and @ueshin
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-47818][CONNECT] Introduce plan cache in SparkConnectPlanner to improve performance of Analyze requests [spark]
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on code in PR #46012:
URL: https://github.com/apache/spark/pull/46012#discussion_r1565802827
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SessionHolder.scala:
##########
@@ -381,6 +405,53 @@ case class SessionHolder(userId: String, sessionId: String, session: SparkSessio
*/
private[connect] val pythonAccumulator: Option[PythonAccumulator] =
Try(session.sparkContext.collectionAccumulator[Array[Byte]]).toOption
+
+ /**
+ * Transform a relation into a logical plan, using the plan cache if enabled.
+ * The plan cache is enable only if `spark.connect.session.planCache.maxSize` is greater than zero
+ * AND `spark.connect.session.planCache.enabled` is true.
+ * @param rel The relation to transform.
+ * @param cachePlan Whether to cache the result logical plan.
+ * @param transform Function to transform the relation into a logical plan.
+ * @return The logical plan.
+ */
+ private[connect] def usePlanCache(rel: proto.Relation, cachePlan: Boolean)(
Review Comment:
> 1, LOCAL_RELATION, its size might be large;
> 2, CACHED_REMOTE_RELATION, it is already cached in dataFrameCache in the SessionHolder;
We use a low default value for `CONNECT_SESSION_PLAN_CACHE_SIZE`, `5` entries to be specific, to prevent the cache from becoming too large. We haven't been too surgical in what kind of plans we cache to keep it as a simple wrapper in aiding common scenarios that may generate some repeated analysis/execute calls.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-47818][CONNECT] Introduce plan cache in SparkConnectPlanner to improve performance of Analyze requests [spark]
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #46012:
URL: https://github.com/apache/spark/pull/46012#discussion_r1566710260
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SessionHolder.scala:
##########
@@ -381,6 +405,53 @@ case class SessionHolder(userId: String, sessionId: String, session: SparkSessio
*/
private[connect] val pythonAccumulator: Option[PythonAccumulator] =
Try(session.sparkContext.collectionAccumulator[Array[Byte]]).toOption
+
+ /**
+ * Transform a relation into a logical plan, using the plan cache if enabled.
+ * The plan cache is enable only if `spark.connect.session.planCache.maxSize` is greater than zero
+ * AND `spark.connect.session.planCache.enabled` is true.
+ * @param rel The relation to transform.
+ * @param cachePlan Whether to cache the result logical plan.
+ * @param transform Function to transform the relation into a logical plan.
+ * @return The logical plan.
+ */
+ private[connect] def usePlanCache(rel: proto.Relation, cachePlan: Boolean)(
Review Comment:
yea, I forgot the new `plan_id`. Then it's fine
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-47818][CONNECT] Introduce plan cache in SparkConnectPlanner to improve performance of Analyze requests [spark]
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #46012:
URL: https://github.com/apache/spark/pull/46012#discussion_r1562547428
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SessionHolder.scala:
##########
@@ -381,6 +405,53 @@ case class SessionHolder(userId: String, sessionId: String, session: SparkSessio
*/
private[connect] val pythonAccumulator: Option[PythonAccumulator] =
Try(session.sparkContext.collectionAccumulator[Array[Byte]]).toOption
+
+ /**
+ * Transform a relation into a logical plan, using the plan cache if enabled.
+ * The plan cache is enable only if `spark.connect.session.planCache.maxSize` is greater than zero
+ * AND `spark.connect.session.planCache.enabled` is true.
+ * @param rel The relation to transform.
+ * @param cachePlan Whether to cache the result logical plan.
+ * @param transform Function to transform the relation into a logical plan.
+ * @return The logical plan.
+ */
+ private[connect] def usePlanCache(rel: proto.Relation, cachePlan: Boolean)(
Review Comment:
I think we may want to exclude some kinds of plans:
1, LOCAL_RELATION, its size might be large;
2, CACHED_REMOTE_RELATION, it is already cached in `dataFrameCache` in the `SessionHolder`;
3, CATALOG, it may be a command like `CREATE_TABLE/DROP_TEMP_VIEW `, should not be skipped.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-47818][CONNECT] Introduce plan cache in SparkConnectPlanner to improve performance of Analyze requests [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #46012:
URL: https://github.com/apache/spark/pull/46012#issuecomment-2058415813
Merged to master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-47818] Introduce plan cache in SparkConnectPlanner to improve performance of Analyze requests [spark]
Posted by "beliefer (via GitHub)" <gi...@apache.org>.
beliefer commented on PR #46012:
URL: https://github.com/apache/spark/pull/46012#issuecomment-2051149021
I submitted https://github.com/apache/spark/pull/43473 half year ago.
It seems not received enough attention.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org