API Reference¶

Reference documentation for the Spark client and related APIs.

Spark Client¶

SparkClient for Kubeflow SDK.

class kubeflow.spark.api.spark_client.SparkClient(backend_config: KubernetesBackendConfig | None = None)[source]¶

Bases: object

Stateless Spark client for Kubeflow.

__init__(backend_config: KubernetesBackendConfig | None = None)[source]¶

Initialize the Spark client.

Parameters:: backend_config (KubernetesBackendConfig | None) – Kubernetes backend configuration. If not provided, the default configuration is used.
Raises:: ValueError – If backend_config is not a KubernetesBackendConfig instance.

Connect to or create a SparkConnect session.

This method supports two modes based on parameters: - Connect mode: When base_url is provided, connects to an existing Spark Connect server - Create mode: When base_url is not provided, creates a new Spark Connect session

Parameters:

base_url (str | None) – Optional URL to existing Spark Connect server (e.g., “sc://server:15002”). If provided, connects to existing server. If None, creates new session.
token (str | None) – Optional authentication token for existing server.
num_executors (int | None) – Number of executor instances (create mode only).
resources_per_executor (dict[str, str] | None) – Resource requirements per executor as dict. Format: {“cpu”: “5”, “memory”: “10Gi”} (create mode only).
spark_conf (dict[str, str] | None) – Spark configuration dictionary (create mode only).
driver (Driver | None) – Driver configuration object (create mode only).
executor (Executor | None) – Executor configuration object (create mode only).
options (list | None) – List of configuration options (create mode only). Use Name option for custom session name.
timeout (int) – Timeout in seconds to wait for session ready.
connect_timeout (int) – Timeout in seconds for SparkSession.getOrCreate() (create mode only).

Returns:

SparkSession connected to Spark (self-managing).

Raises:

ValueError – If base_url is invalid or the provided resource configuration is invalid.
TimeoutError – If creating a Spark Connect session or connecting to it times out.
RuntimeError – If the Spark Connect session cannot be created or connected to.

Note

Server port defaults to 15002 (Spark Connect gRPC). PySpark and server Spark major.minor should match; see constants and pyproject.toml [spark].

list_sessions() → list[SparkConnectInfo][source]¶

List SparkConnect sessions.

Returns:: List of SparkConnectInfo objects.

get_session(name: str) → SparkConnectInfo[source]¶

Get information about a SparkConnect session.

Parameters:: name (str) – Name of the SparkConnect session.
Returns:: SparkConnectInfo containing information about the SparkConnect session.

delete_session(name: str) → None[source]¶

Delete a SparkConnect session.

Parameters:: name (str) – Name of the SparkConnect session to delete.

get_session_logs(name: str, follow: bool = False) → Iterator[str][source]¶

Get logs from a SparkConnect session.

Parameters:

name (str) – Name of the SparkConnect session.
follow (bool) – Whether to stream logs continuously.

Returns:

Iterator of log lines from the SparkConnect driver pod.

Submit a batch Spark job.

This method supports two job types:

FileJob: Submit a Spark application referenced by a local

or remote file source. - FuncJob: Submit a Python function as a Spark batch job.

Parameters:

job (FileJob | FuncJob) – Job definition describing the workload to execute. Supports either FileJob or FuncJob.
num_executors (int | None) – Number of executor instances.
resources_per_executor (dict[str, str] | None) – Resource requirements per executor. Format: {"cpu": "5", "memory": "10Gi"}.
spark_conf (dict[str, str] | None) – Spark configuration properties.
options (list | None) – List of additional Spark configuration options.

Raises:

ValueError – If unsupported Phase 1 features are requested or the job definition is invalid.
TypeError – If the job type is invalid.
NotImplementedError – If unsupported features are requested.

get_job(name: str) → SparkJob[source]¶

Get information about a Spark job.

Parameters:: name (str) – Name of the Spark job.
Returns:: SparkJob containing information about the Spark job.

list_jobs(status: set[SparkJobStatus] | None = None) → list[SparkJob][source]¶

List Spark jobs.

Parameters:: status (set[SparkJobStatus] | None) – Optional set of job statuses to filter the returned jobs.
Returns:: List of SparkJob objects.

delete_job(name: str) → None[source]¶

Delete a Spark job.

Parameters:: name (str) – Name of the Spark job to delete.

wait_for_job_status(name: str, status: set[SparkJobStatus] = {SparkJobStatus.COMPLETED}, timeout: int = 600, polling_interval: int = 2) → SparkJob[source]¶

Wait for a Spark job to reach one of the target states. :type name: str :param name: Job name. :type status: set[SparkJobStatus] :param status: Target job state(s). :type timeout: int :param timeout: Maximum wait time in seconds. Default 600. :type polling_interval: int :param polling_interval: Time in seconds between status checks.

Returns:: Spark job information after reaching one of the target statuses.
Raises:: ValueError – If the polling interval or timeout values are invalid.

get_job_logs(name: str, follow: bool = False) → Iterator[str][source]¶

Get logs from a Spark job.

Parameters:

name (str) – Spark job name.
follow (bool) – Whether to stream logs in realtime.

Returns:

Iterator of log lines.

Types¶

Types for Kubeflow Spark SDK.

class kubeflow.spark.types.types.SparkConnectState(value)[source]¶

Bases: str, Enum

State of a SparkConnect session.

PROVISIONING = 'Provisioning'¶

READY = 'Ready'¶

RUNNING = 'Running'¶

NOT_READY = 'NotReady'¶

FAILED = 'Failed'¶

class kubeflow.spark.types.types.SparkConnectInfo(name: str, namespace: str, state: SparkConnectState, driver_pod_name: str | None = None, pod_ip: str | None = None, service_name: str | None = None, creation_timestamp: datetime | None = None) → None[source]¶

Bases: object

Information about a SparkConnect session.

Parameters:

name (str) – Name of the SparkConnect session.
namespace (str) – Kubernetes namespace. Included in SparkConnectInfo for standalone usage and passing info between components without requiring SparkClient context.
state (SparkConnectState) – Current state of the session.
driver_pod_name (str | None) – Name of the driver pod.
pod_ip (str | None) – IP address of the server pod.
service_name (str | None) – Name of the Kubernetes service.
creation_timestamp (datetime | None) – Timestamp when the session was created.

name: str¶

namespace: str¶

state: SparkConnectState¶

driver_pod_name: str | None = None¶

pod_ip: str | None = None¶

service_name: str | None = None¶

creation_timestamp: datetime | None = None¶

class kubeflow.spark.types.types.Driver(image: str | None = None, resources: dict[str, str] | None = None, java_options: str | None = None, service_account: str | None = None) → None[source]¶

Bases: object

Driver configuration for Spark Connect session.

The Driver configuration allows fine-grained control over the Spark driver pod. All fields are optional, with sensible defaults applied by the backend.

Parameters:

image (str | None) – Custom container image for the driver.
resources (dict[str, str] | None) – Resource requirements as dict (e.g., {“cpu”: “2”, “memory”: “4Gi”}).
java_options (str | None) – JVM options for the driver (e.g., “-Xmx4g -XX:+UseG1GC”).
service_account (str | None) – Kubernetes service account name for RBAC.

Example:

driver = Driver(
    resources={
        "cpu": "4",
        "memory": "8Gi",
    },
    service_account="spark-driver-prod",
)

Note

The resources dict is extensible - any valid Kubernetes resource name is supported. This design allows future resource types without API changes.

image: str | None = None¶

resources: dict[str, str] | None = None¶

java_options: str | None = None¶

service_account: str | None = None¶

class kubeflow.spark.types.types.Executor(num_instances: int | None = None, resources_per_executor: dict[str, str] | None = None, java_options: str | None = None) → None[source]¶

Bases: object

Executor configuration for Spark Connect session.

The Executor configuration controls the worker pods that execute Spark tasks. All fields are optional, with sensible defaults applied by the backend.

Parameters:

num_instances (int | None) – Number of executor instances (pods).
resources_per_executor (dict[str, str] | None) – Resource requirements per executor as dict (e.g., {“cpu”: “4”, “memory”: “8Gi”}).
java_options (str | None) – JVM options for executors (e.g., “-Xmx28g -XX:+UseG1GC”).

Example:

executor = Executor(
    num_instances=20,
    resources_per_executor={
        "cpu": "8",
        "memory": "32Gi",
    },
)

Note

The resources_per_executor dict is extensible - any valid Kubernetes resource name is supported. This design allows future resource types without API changes.

num_instances: int | None = None¶

resources_per_executor: dict[str, str] | None = None¶

java_options: str | None = None¶

class kubeflow.spark.types.types.SparkJobStatus(value)[source]¶

Bases: str, Enum

State of a Spark batch job.

CREATED = 'Created'¶

RUNNING = 'Running'¶

COMPLETED = 'Completed'¶

FAILED = 'Failed'¶

classmethod from_operator_state(raw_state: str | None) → SparkJobStatus[source]¶

Map a SparkApplication state to a SparkJobStatus.

Parameters:: raw_state (str | None) – SparkApplication applicationState.state value.
Returns:: Corresponding SparkJobStatus.

Note

Unknown SparkApplication states default to FAILED so newly introduced operator states are handled conservatively.

class kubeflow.spark.types.types.SparkJob(name: str, namespace: str, status: SparkJobStatus | None = None, creation_timestamp: datetime | None = None, num_executors: int | None = None, driver_pod_name: str | None = None) → None[source]¶

Bases: object

Information about a Spark batch job.

Parameters:

name (str) – Name of the SparkApplication.
namespace (str) – Kubernetes namespace containing the SparkApplication. Included in SparkJob for standalone usage and passing job information between components without requiring SparkClient context.
status (SparkJobStatus | None) – Current state of the Spark batch job.
creation_timestamp (datetime | None) – Timestamp when the SparkApplication was created.
num_executors (int | None) – Number of configured Spark executor instances.
driver_pod_name (str | None) – Name of the Spark driver pod, if available.

name: str¶

namespace: str¶

status: SparkJobStatus | None = None¶

creation_timestamp: datetime | None = None¶

num_executors: int | None = None¶

driver_pod_name: str | None = None¶

class kubeflow.spark.types.types.FileJob(file_source: str, args: list[str] | None = None) → None[source]¶

Bases: object

Spark application referenced by a local or remote file source.

Parameters:

file_source (str) – Path or URI of the Spark application. Supports local paths available to the Spark cluster as well as remote URIs such as s3a://, gs://, hdfs:// and https://.
args (list[str] | None) – Optional command-line arguments passed to the application.

file_source: str¶

args: list[str] | None = None¶

class kubeflow.spark.types.types.FuncJob(func: Callable, func_args: dict[str, Any] | None = None) → None[source]¶

Bases: object

Function-based Spark application.

The provided function must be self-contained. Any required imports should be placed inside the function body. Module-level globals, closures, and decorated functions are not supported.

Parameters:

func (Callable) – Python function executed as a Spark batch job.
func_args (dict[str, Any] | None) – Optional keyword arguments passed to the function.

func: Callable¶

func_args: dict[str, Any] | None = None¶

Kubernetes Options¶

Options for advanced Spark configuration (KEP-107 lines 180-192).

The options pattern provides extensibility for advanced Kubernetes configurations without polluting the main API. Future option types can be added without breaking changes.

This follows the same callable pattern as kubeflow.trainer.options for SDK consistency.

class kubeflow.spark.options.kubernetes.Labels(labels: dict[str, str]) → None[source]¶

Bases: object

Add Kubernetes labels to Spark resources (.metadata.labels).

Labels are key-value pairs attached to Kubernetes resources for organization, selection, and grouping.

Supported backends:

Kubernetes

Parameters:: labels (dict[str, str]) – Dictionary of label key-value pairs.

Example:

options = [
    Labels(
        {
            "app": "spark",
            "team": "data-eng",
        }
    ),
]
spark = client.connect(..., options=options)

labels: dict[str, str]¶

__call__(resource: SparkV1alpha1SparkConnect | SparkV1beta2SparkApplication, backend: RuntimeBackend) → None[source]¶

Apply labels to the Spark resource.

Parameters:

resource (SparkV1alpha1SparkConnect | SparkV1beta2SparkApplication) – Spark resource to modify.
backend (RuntimeBackend) – Backend instance for validation.

Raises:

ValueError – If backend does not support labels.

class kubeflow.spark.options.kubernetes.Annotations(annotations: dict[str, str]) → None[source]¶

Bases: object

Add Kubernetes annotations to Spark resources (.metadata.annotations).

Annotations store non-identifying metadata that can be used by tools, libraries, or for documentation purposes.

Supported backends:

Kubernetes

Parameters:: annotations (dict[str, str]) – Dictionary of annotation key-value pairs.

Example:

options = [
    Annotations(
        {
            "description": "Daily ETL pipeline",
            "owner": "data-team@company.com",
        }
    ),
]
spark = client.connect(..., options=options)

annotations: dict[str, str]¶

__call__(resource: SparkV1alpha1SparkConnect | SparkV1beta2SparkApplication, backend: RuntimeBackend) → None[source]¶

Apply annotations to the Spark resource.

Parameters:

resource (SparkV1alpha1SparkConnect | SparkV1beta2SparkApplication) – Spark resource to modify.
backend (RuntimeBackend) – Backend instance for validation.

Raises:

ValueError – If backend does not support annotations.

class kubeflow.spark.options.kubernetes.NodeSelector(selectors: dict[str, str]) → None[source]¶

Bases: object

Add node selector constraints to Spark pods.

Node selectors constrain pod scheduling to nodes with matching labels. Applied to both driver and executor pods.

Supported backends:

Kubernetes

Parameters:: selectors (dict[str, str]) – Dictionary of node label key-value pairs.

Example:

options = [
    NodeSelector(
        {
            "node-type": "spark",
            "gpu": "true",
        }
    ),
]
spark = client.connect(..., options=options)

selectors: dict[str, str]¶

__call__(resource: SparkV1alpha1SparkConnect | SparkV1beta2SparkApplication, backend: RuntimeBackend) → None[source]¶

Apply node selector constraints to the Spark resource.

Parameters:

resource (SparkV1alpha1SparkConnect | SparkV1beta2SparkApplication) – Spark resource to modify.
backend (RuntimeBackend) – Backend instance for validation.

Raises:

ValueError – If backend does not support node selectors.
TypeError – If the resource is not a supported Spark resource.

class kubeflow.spark.options.kubernetes.Toleration(key: str, operator: str = 'Equal', value: str = '', effect: str = 'NoSchedule') → None[source]¶

Bases: object

Add toleration to Spark pods for node taints.

Tolerations allow pods to schedule onto nodes with matching taints. Applied to both driver and executor pods.

Supported backends:

Kubernetes

Parameters:

key (str) – Taint key to tolerate.
operator (str) – Operator (Equal or Exists).
value (str) – Taint value (if operator is Equal).
effect (str) – Taint effect (NoSchedule, PreferNoSchedule, or NoExecute).

Example:

options = [
    Toleration(
        key="spark-workload",
        operator="Equal",
        value="true",
        effect="NoSchedule",
    ),
]
spark = client.connect(..., options=options)

key: str¶

operator: str = 'Equal'¶

value: str = ''¶

effect: str = 'NoSchedule'¶

__call__(resource: SparkV1alpha1SparkConnect | SparkV1beta2SparkApplication, backend: RuntimeBackend) → None[source]¶

Apply toleration to the Spark resource.

Parameters:

resource (SparkV1alpha1SparkConnect | SparkV1beta2SparkApplication) – Spark resource to modify.
backend (RuntimeBackend) – Backend instance for validation.

Raises:

ValueError – If backend does not support tolerations.
TypeError – If the resource is not a supported Spark resource.

class kubeflow.spark.options.kubernetes.Name(name: str) → None[source]¶

Bases: object

Set a custom name for the Spark resource.

This option sets the Kubernetes resource name.

If not provided, a name is automatically generated: - Spark Connect sessions: spark-connect-{uuid} - Spark batch jobs: spark-job-{uuid}

The session name must follow DNS-1123 subdomain rules: - Lowercase alphanumeric characters, ‘-’, or ‘.’ - Start and end with alphanumeric character - Maximum 253 characters

Supported backends:

Kubernetes

Parameters:: name (str) – Custom name for the session. Must be a valid Kubernetes resource name.

Example:

from kubeflow.spark import SparkClient
from kubeflow.spark.options import Name

client = SparkClient()

# With explicit name
spark = client.connect(options=[Name("my-custom-session")])

# Auto-generated name
spark = client.connect()  # Creates "spark-connect-a1b2c3d4"

name: str¶

__call__(resource: SparkV1alpha1SparkConnect | SparkV1beta2SparkApplication, backend: RuntimeBackend) → None[source]¶

Apply custom name to the Spark resource metadata.

Parameters:

resource (SparkV1alpha1SparkConnect | SparkV1beta2SparkApplication) – Spark resource to modify.
backend (RuntimeBackend) – Backend instance for validation.

Raises:

ValueError – If backend does not support custom names.