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)

Bases: object

Stateless Spark client for Kubeflow.

__init__(backend_config: KubernetesBackendConfig | None = None)

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(base_url: str | None = None, token: str | None = None, num_executors: int | None = None, resources_per_executor: dict[str, str] | None = None, spark_conf: dict[str, str] | None = None, driver: Driver | None = None, executor: Executor | None = None, options: list | None = None, timeout: int = 300, connect_timeout: int = 120) → SparkSession

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]

List SparkConnect sessions.

Returns:

List of SparkConnectInfo objects.

get_session(name: str) → SparkConnectInfo

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

Delete a SparkConnect session.

Parameters:

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

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

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_job(job: FileJob | FuncJob, num_executors: int | None = None, resources_per_executor: dict[str, str] | None = None, spark_conf: dict[str, str] | None = None, options: list | None = None) → str

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

• 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

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]

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

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

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]

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.