Spark Jobs - oleander

`get_spark_cluster(name_or_options)`

Look up Spark cluster details before submitting a job. Use oleander for the built-in managed cluster or the name of a registered external cluster.

cluster = await client.get_spark_cluster("emr-prod")
print(cluster.type, cluster.properties)

Return shape

Field	Type	Description
`name`	`str`	Cluster name
`type`	`SparkClusterType`	`oleander`, `emr-serverless`, or `glue`
`properties`	`Any`	Cluster-specific configuration

`list_spark_jobs(options?)`

List your uploaded Spark job artifacts with pagination.

jobs = await client.list_spark_jobs()
for artifact in jobs.artifacts:
    print(artifact.name, artifact.entrypoint, artifact.status)

from oleander_sdk import ListSparkJobsOptions

offset = 0
all_artifacts = []
while True:
    page = await client.list_spark_jobs(
        ListSparkJobsOptions(limit=50, offset=offset)
    )
    all_artifacts.extend(page.artifacts)
    if not page.has_more:
        break
    offset += 50

Parameters

options.limit

int

default:"20"

Number of artifacts to return per page. Must be greater than 0.

options.offset

int

default:"0"

Number of artifacts to skip for pagination. Must be 0 or greater.

Return type: `ListSparkJobsResult`

Field	Type	Description
`artifacts`	`list[SparkArtifactSummary]`	Artifacts for the current page
`has_more`	`bool`	Whether more artifacts are available

Each SparkArtifactSummary has:

Field	Type	Description
`id`	`str`	Artifact ID
`entrypoint`	`str`	Entrypoint script or class reference
`name`	`str`	Artifact name
`version`	`int`	Artifact version
`language`	`str`	Artifact language
`main_class`	`Optional[str]`	Main class for JVM jobs
`py_files`	`Optional[str]`	Python dependencies archive
`virtualenv`	`Optional[str]`	Virtualenv archive
`status`	`str`	Artifact status
`created_at`	`Optional[str]`	ISO timestamp when created
`updated_at`	`Optional[str]`	ISO timestamp when last updated

`submit_spark_job(options)`

Submit a Spark job for execution. cluster defaults to oleander. For oleander-managed Spark, entrypoint is the uploaded script name. For external clusters, entrypoint is cluster-specific, such as an S3 URI for EMR Serverless or a Glue job name for Glue. The canonical field is entrypoint. script_name is still accepted as a compatibility alias.

from oleander_sdk import SparkJobSubmitOptions

result = await client.submit_spark_job(SparkJobSubmitOptions(
    namespace="my-namespace",
    name="daily-etl",
    entrypoint="etl_pipeline.py",
    args=["--date", "2026-03-11"],
    executor_numbers=4,
))

print(result.run_id)
run = await client.get_run(result.run_id)

Common options

Option	Type	Default	Description
`cluster`	`str`	`"oleander"`	Managed cluster or registered cluster name
`namespace`	`str`		Job namespace
`name`	`str`		Job name
`entrypoint`	`str`		Script name, S3 URI, or Glue job name depending on cluster
`args`	`list[str]`	`[]`	Entrypoint arguments
`spark_conf`	`list[str]`	`[]`	Spark configuration values
`packages`	`list[str]`	`[]`	Extra package coordinates
`job_tags`	`list[str]`	`[]`	Tags applied to the job
`run_tags`	`list[str]`	`[]`	Tags applied to this run

Oleander-managed options

Option	Type	Default	Description
`driver_machine_type`	`SparkMachineType`	`spark.1.b`	Driver machine type
`executor_machine_type`	`SparkMachineType`	`spark.1.b`	Executor machine type
`executor_numbers`	`int`	`2`	Number of executors, from 1 to 20

EMR Serverless options

Option	Type	Description
`py_files`	`str`	Zip archive of Python dependencies
`virtualenv`	`str`	Virtualenv archive for Python jobs
`main_class`	`str`	Main class for JVM jobs
`execution_iam_policy`	`str`	IAM policy applied to execution

Glue options

Option	Type	Default	Description
`worker_type`	`GlueWorkerType`		Glue worker type
`number_of_workers`	`int`	`1`	Number of Glue workers
`enable_auto_scaling`	`Optional[bool]`		Enable Glue auto scaling
`timeout_minutes`	`Optional[int]`		Timeout in minutes
`execution_class`	`Optional[str]`		Glue execution class such as `STANDARD` or `FLEX`
`execution_iam_policy`	`Optional[str]`		IAM policy applied to execution

For Glue jobs, args are converted into key-value pairs. Pass them as alternating entries such as ["--source", "s3://bucket/input", "--target", "s3://bucket/output"].

External cluster example

result = await client.submit_spark_job(SparkJobSubmitOptions(
    cluster="emr-prod",
    namespace="finance",
    name="daily-etl",
    entrypoint="s3://my-bucket/jobs/etl_pipeline.py",
    args=["--date", "2026-03-11"],
    py_files="s3://my-bucket/jobs/deps.zip",
    packages=["org.example:my-lib:1.0.0"],
))

Return type: `SparkJobRun`

Field	Type	Description
`run_id`	`str`	The ID of the submitted run

Machine types

The SparkMachineType enum covers compute-optimized (c), balanced (b), and memory-optimized (m) options:

Type	vCPUs	Category
`spark.1.c` / `spark.1.b` / `spark.1.m`	1	Compute / Balanced / Memory
`spark.2.c` / `spark.2.b` / `spark.2.m`	2	Compute / Balanced / Memory
`spark.4.c` / `spark.4.b` / `spark.4.m`	4	Compute / Balanced / Memory
`spark.8.c` / `spark.8.b` / `spark.8.m`	8	Compute / Balanced / Memory
`spark.16.c` / `spark.16.b` / `spark.16.m`	16	Compute / Balanced / Memory

`submit_spark_job_and_wait(options)`

Submit a Spark job and poll until it reaches a terminal state (COMPLETE, FAIL, or ABORT). Raises TimeoutError if the timeout is exceeded.

from oleander_sdk import SubmitSparkJobAndWaitOptions

result = await client.submit_spark_job_and_wait(SubmitSparkJobAndWaitOptions(
    namespace="my-namespace",
    name="daily-etl",
    entrypoint="etl_pipeline.py",
    poll_interval_ms=5000,
    timeout_ms=300000,
))

if result.state == "COMPLETE":
    elapsed = result.run.duration
    # proceed with downstream work ...
else:
    raise Exception(f"Run {result.run_id} ended with state: {result.state}")

Accepts all submit_spark_job options plus:

options.poll_interval_ms

int

default:"10000"

Milliseconds between status polls. Must be greater than 0.

options.timeout_ms

int

default:"600000"

Maximum time to wait in milliseconds before raising TimeoutError.

Return type: `SubmitAndWaitResult`

Field	Type	Description
`run_id`	`str`	The ID of the submitted run
`state`	`str`	Terminal state (`COMPLETE`, `FAIL`, or `ABORT`)
`run`	`RunResponse`	Full run details

`get_run(run_id)`

Get the current status of a run. Use this to poll a job submitted with submit_spark_job().

run = await client.get_run(run_id)

if run.state == "COMPLETE":
    duration = run.duration
    job_name = run.job.name
    # handle completion ...
elif run.state == "FAIL":
    error = run.error
    # handle failure ...

Return type: `RunResponse`

Field	Type	Description
`id`	`str`	Run ID
`state`	`Optional[str]`	Current state
`started_at`	`Optional[str]`	ISO timestamp when the run started
`queued_at`	`Optional[str]`	ISO timestamp when the run was queued
`scheduled_at`	`Optional[str]`	ISO timestamp when the run was scheduled
`ended_at`	`Optional[str]`	ISO timestamp when the run ended
`duration`	`Optional[float]`	Run duration in seconds
`error`	`Optional[Any]`	Error details if the run failed
`tags`	`list[RunTag]`	List of tags with `key`, `value`, and optional `source`
`job`	`RunJobInfo`	Job info with `id`, `name`, `namespace`
`pipeline`	`RunPipelineInfo`	Pipeline info with `id`, `name`, `namespace`

​get_spark_cluster(name_or_options)

​Return shape

​list_spark_jobs(options?)

​Parameters

​Return type: ListSparkJobsResult

​submit_spark_job(options)

​Common options

​Oleander-managed options

​EMR Serverless options

​Glue options

​External cluster example

​Return type: SparkJobRun

​Machine types

​submit_spark_job_and_wait(options)

​Return type: SubmitAndWaitResult

​get_run(run_id)

​Return type: RunResponse

`get_spark_cluster(name_or_options)`

Return shape

`list_spark_jobs(options?)`

Parameters

Return type: `ListSparkJobsResult`

`submit_spark_job(options)`

Common options

Oleander-managed options

EMR Serverless options

Glue options

External cluster example

Return type: `SparkJobRun`

Machine types

`submit_spark_job_and_wait(options)`

Return type: `SubmitAndWaitResult`

`get_run(run_id)`

Return type: `RunResponse`