[amoghrajesh]

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.

What changes were proposed in this pull request?

Adds SparkSession.getOrCreate(path, create) to Spark Classic, in Scala and
PySpark.

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

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

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:

>>> 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:

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:

var calls = 0
def create() = {
  calls += 1
  println(">>> computing")
  Thread.sleep(8000)
  spark.range(1000).toDF()
}

Running this:

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> 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

def job(date: String) = spark.getOrCreate(
  s"/tmp/sdaily/$date",
  () => { println(s">>> computing $date"); Thread.sleep(5000); spark.range(100).toDF() })

Result:

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, is that an acceptable issue?
• Cleanup: nothing deletes saved intermediates. Should that be documented?
• Concurrency: two runs against the same fresh can path both compute and overwrite.
• Spark 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