amoghrajesh opened a new pull request, #60000:
URL: https://github.com/apache/spark/pull/60000
Given a path, load the dataset if it is already there; otherwise compute it,
store it as Parquet, and return it. Lets a rerun skip work an earlier run
finished.
Classic only. Completeness is tracked with a marker file instead of reusing
`_SUCCESS`, which can be disabled or left by an unrelated writer.
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR:
https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g.,
'[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a
faster review.
7. If you want to add a new configuration, please read the guideline first
for naming configurations in
'common/utils/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the
guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
Adds `SparkSession.getOrCreate(path, create)` to Spark Classic, in Scala and
PySpark.
```python
df = spark.getOrCreate(path, lambda: expensive_transform(src))
```
First call computes and saves. Later calls load and skip the work. A marker
file records completeness, so a path left by an interrupted write is recomputed.
Classic only, `DataFrame` not `Dataset[T]`, Parquet hardcoded, caller
supplies the path.
### Why are the changes needed?
A batch job that dies in a late stage reruns from the start. Under an
orchestrator, users pay for the same stages twice, usually after a spot reclaim
where the same code would have worked.
People hand-roll this today and get it wrong quietly: a path-exists check
returns partial data, and a `_SUCCESS` check returns an unrelated job's data.
Both report success.
Dev list: [Thoughts around getOrCreate for Dataset /
DataFrame](https://lists.apache.org/thread.html/7mj5rwnylksnrvjocywc6sfxnrhl7c59)
### Does this PR introduce _any_ user-facing change?
Yes, a new public method. No change to existing APIs.
### How was this patch tested?
New suites in Scala and PySpark, three tests each: computes on first call,
loads instead of recomputing, recomputes when the marker is absent.
#### Pyspark
Using this function for the test - imitates a function run by daily pipeline
that passes in the date
```python
import time, os, shutil
def job(date):
def create():
print(">>> computing", date)
time.sleep(5)
return spark.range(100)
return spark.getOrCreate(f"/tmp/daily/{date}", create)
```
Result:
```shell
>>> job("2026-11-01").count(); job("2026-11-02").count()
>>> computing 2026-11-01
>>> computing 2026-11-02 # separate folders,
no reuse
100
>>> job("2026-11-01").count() # a retry
100 # instant, no
recompute
>>> [job(d).count() for d in ["2026-11-03", ..., "2026-11-07"]]
>>> computing 2026-11-03 ... 2026-11-07
>>> sorted(os.listdir("/tmp/daily"))
['2026-11-01', '2026-11-02', '2026-11-03', '2026-11-04', '2026-11-05',
'2026-11-06', '2026-11-07']
9926 bytes # nothing cleaned up
```
Listing the files:
```shell
amoghrajesh.desai …/spark SPARK-60775 v26.0.1 🅼 ♥ 16:16
ls -lr /tmp/daily
total 0
drwxr-xr-x@ 10 amoghrajesh.desai wheel 320 Nov 1 16:12 2026-11-07
drwxr-xr-x@ 10 amoghrajesh.desai wheel 320 Nov 1 16:12 2026-11-06
drwxr-xr-x@ 10 amoghrajesh.desai wheel 320 Nov 1 16:12 2026-11-05
drwxr-xr-x@ 10 amoghrajesh.desai wheel 320 Nov 1 16:11 2026-11-04
drwxr-xr-x@ 10 amoghrajesh.desai wheel 320 Nov 1 16:11 2026-11-03
drwxr-xr-x@ 10 amoghrajesh.desai wheel 320 Nov 1 16:11 2026-11-02
drwxr-xr-x@ 10 amoghrajesh.desai wheel 320 Nov 1 16:10 2026-11-01
```
#### Scala
Running same / similar case with scala. Run the shell first:
`./bin/spark-shell`
Lets use this block:
```scala
var calls = 0
def create() = {
calls += 1
println(">>> computing")
Thread.sleep(8000)
spark.range(1000).toDF()
}
```
Running this:
```scala
scala> val t = System.currentTimeMillis; val n =
spark.getOrCreate("/tmp/sgoc1", () => create()).count(); println(s"count $n |
calls $calls | secs ${(System.currentTimeMillis-t)/1000.0}")
|
>>> computing
count 1000 | calls 1 | secs 10.4
val t: Long = 1793531061641
val n: Long = 1000
```
That notes the time, calls getOrCreate on an empty path, counts the rows,
then prints how long it took and how many times the work ran.
Then this:
```scala
scala> val t = System.currentTimeMillis; val n =
spark.getOrCreate("/tmp/sgoc1", () => create()).count(); println(s"count $n |
calls $calls | secs ${(System.currentTimeMillis-t)/1000.0}")
|
>>> computing
count 1000 | calls 1 | secs 10.4
val t: Long = 1793531061641
val n: Long = 1000
```
##### Daily pipeline similar to pyspark
```scala
def job(date: String) = spark.getOrCreate(
s"/tmp/sdaily/$date",
() => { println(s">>> computing $date"); Thread.sleep(5000);
spark.range(100).toDF() })
```
Result:
```shell
scala> job("2026-11-01").count(); job("2026-11-02").count()
|
>>> computing 2026-11-01
>>> computing 2026-11-02
val res0: Long = 100
```
### Open questions
- Staleness: if an input changes but the saved copy does not, the stale copy
is
returned. No way to force recomputation yet.
- Cleanup: nothing deletes saved intermediates.
- Concurrency: two runs against the same fresh path both compute and
overwrite.
- Connect support, and a format argument, left as follow-ups.
### Was this patch authored or co-authored using generative AI tooling?
Yes, Claude Code - sonnet
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
