[SPARK-47323][K8S] Support custom executor log urls #14
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
EnricoMi
changed the title
EnricoMi
force-pushed
the
k8s-custom-executor-log-url
branch
from
27af189 to
239fb2f
Compare
EnricoMi
force-pushed
the
k8s-custom-executor-log-url
branch
from
a465e9b to
b9ea61b
Compare
EnricoMi
force-pushed
the
k8s-custom-executor-log-url
branch
from
80070ef to
2f896c8
Compare
github-actions
bot
added
CORE
SQL
PYTHON
INFRA
BUILD
MLLIB
STRUCTURED STREAMING
EXAMPLES
ML
PANDAS API ON SPARK
YARN
AVRO
DSTREAM
R
CONNECT
PROTOBUF
GRAPHX
labels
|
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable. |
EnricoMi
force-pushed
the
k8s-custom-executor-log-url
branch
from
2f896c8 to
315a0bb
Compare
github-actions
bot
removed
CORE
SQL
PYTHON
INFRA
BUILD
MLLIB
STRUCTURED STREAMING
EXAMPLES
ML
PANDAS API ON SPARK
YARN
AVRO
DSTREAM
R
CONNECT
PROTOBUF
GRAPHX
labels
EnricoMi
pushed a commit
that referenced
this pull request
Sep 2, 2025
…onicalized expressions
### What changes were proposed in this pull request?
Make PullOutNonDeterministic use canonicalized expressions to dedup group and aggregate expressions. This affects pyspark udfs in particular. Example:
```
from pyspark.sql.functions import col, avg, udf
pythonUDF = udf(lambda x: x).asNondeterministic()
spark.range(10)\
.selectExpr("id", "id % 3 as value")\
.groupBy(pythonUDF(col("value")))\
.agg(avg("id"), pythonUDF(col("value")))\
.explain(extended=True)
```
Currently results in a plan like this:
```
Aggregate [_nondeterministic#15](#15), [_nondeterministic#15 AS dummyNondeterministicUDF(value)#12, avg(id#0L) AS avg(id)#13, dummyNondeterministicUDF(value#6L)#8 AS dummyNondeterministicUDF(value)#14](#15%20AS%20dummyNondeterministicUDF(value)#12,%20avg(id#0L)%20AS%20avg(id)#13,%20dummyNondeterministicUDF(value#6L)#8%20AS%20dummyNondeterministicUDF(value)#14)
+- Project [id#0L, value#6L, dummyNondeterministicUDF(value#6L)#7 AS _nondeterministic#15](#0L,%20value#6L,%20dummyNondeterministicUDF(value#6L)#7%20AS%20_nondeterministic#15)
+- Project [id#0L, (id#0L % cast(3 as bigint)) AS value#6L](#0L,%20(id#0L%20%%20cast(3%20as%20bigint))%20AS%20value#6L)
+- Range (0, 10, step=1, splits=Some(2))
```
and then it throws:
```
[[MISSING_AGGREGATION] The non-aggregating expression "value" is based on columns which are not participating in the GROUP BY clause. Add the columns or the expression to the GROUP BY, aggregate the expression, or use "any_value(value)" if you do not care which of the values within a group is returned. SQLSTATE: 42803
```
- how canonicalized fixes this:
- nondeterministic PythonUDF expressions always have distinct resultIds per udf
- The fix is to canonicalize the expressions when matching. Canonicalized means that we're setting the resultIds to -1, allowing us to dedup the PythonUDF expressions.
- for deterministic UDFs, this rule does not apply and "Post Analysis" batch extracts and deduplicates the expressions, as expected
### Why are the changes needed?
- the output of the query with the fix applied still makes sense - the nondeterministic UDF is invoked only once, in the project.
### Does this PR introduce _any_ user-facing change?
Yes, it's additive, it enables queries to run that previously threw errors.
### How was this patch tested?
- added unit test
### Was this patch authored or co-authored using generative AI tooling?
No
Closes apache#52061 from benrobby/adhoc-fix-pull-out-nondeterministic.
Authored-by: Ben Hurdelhey <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
Make Kubernetes resource manager support existing config
spark.ui.custom.executor.log.url.Allow for
Supports these variables:
APP_ID: The unique application idEXECUTOR_ID: The executor id (a positive integer larger than zero)HOSTNAME: The name of the host where the executor runsKUBERNETES_NAMESPACE: The namespace where the executor pods runKUBERNETES_POD_NAME: The name of the pod that contains the executorFILE_NAME: The name of the log, which is always"log"Why are the changes needed?
Running Spark on Kubernetes requires persisting the logs elsewhere. Having the Spark UI link to those logs is very useful. This is currently only supported by YARN.
Does this PR introduce any user-facing change?
Spark UI provides links to logs when run on Kubernetes.
How was this patch tested?
Unit test and manually tested on minikube K8S cluster.
Was this patch authored or co-authored using generative AI tooling?
No