spark_on_k8s.connect_server

See the reference overview for supported entry points and task guides. A module may also contain helpers outside the documented public API.

ConnectServerConfig dataclass

Explicit inputs for one private, single-driver Spark Connect server.

The image must contain the specified Connect server JAR and a Spark entrypoint. This configuration creates neither namespaces nor service accounts. Its plaintext ClusterIP service requires a trusted network or a separately managed gateway.

Pod placement and credentials reference existing Kubernetes objects only: image_pull_secrets names Secrets used for the driver and executors; driver_env_from_secrets and executor_env_from_secrets map environment variable names to SECRET:KEY references, resolved by the kubelet (this helper never reads Secret values); node selectors map node label keys to values; and tolerations are Kubernetes toleration mappings (key, operator, value, effect, tolerationSeconds).

An optional max_runtime (seconds) stops the server's driver even if no client deletes it.

Source code in spark_on_k8s/connect_server.py

@dataclass(frozen=True, kw_only=True)
class ConnectServerConfig:
    """Explicit inputs for one private, single-driver Spark Connect server.

    The image must contain the specified Connect server JAR and a Spark entrypoint.
    This configuration creates neither namespaces nor service accounts. Its plaintext
    ClusterIP service requires a trusted network or a separately managed gateway.

    Pod placement and credentials reference existing Kubernetes objects only:
    ``image_pull_secrets`` names Secrets used for the driver and executors;
    ``driver_env_from_secrets`` and ``executor_env_from_secrets`` map environment
    variable names to ``SECRET:KEY`` references, resolved by the kubelet (this helper
    never reads Secret values); node selectors map node label keys to values; and
    tolerations are Kubernetes toleration mappings (``key``, ``operator``, ``value``,
    ``effect``, ``tolerationSeconds``).

    An optional max_runtime (seconds) stops the server's driver even if no client deletes it.
    """

    image: str
    server_jar: str
    namespace: str
    service_account: str
    app_name: str = "spark-connect"
    port: int = 15002
    image_pull_policy: str = "IfNotPresent"
    driver_resources: PodResources = field(default_factory=PodResources)
    executor_resources: PodResources = field(default_factory=PodResources)
    executor_instances: ExecutorInstances = field(default_factory=ExecutorInstances)
    spark_conf: dict[str, str] = field(default_factory=dict)
    packages: list[str] = field(default_factory=list)
    driver_node_selector: dict[str, str] = field(default_factory=dict)
    executor_node_selector: dict[str, str] = field(default_factory=dict)
    driver_tolerations: list[dict[str, Any]] = field(default_factory=list)
    executor_tolerations: list[dict[str, Any]] = field(default_factory=list)
    image_pull_secrets: list[str] = field(default_factory=list)
    driver_env_from_secrets: dict[str, str] = field(default_factory=dict)
    executor_env_from_secrets: dict[str, str] = field(default_factory=dict)
    max_runtime: int | None = None

    def __post_init__(self):
        if not isinstance(self.image, str) or not self.image or any(c.isspace() for c in self.image):
            raise ValueError("image must be an explicit nonblank container image")
        if (
            not isinstance(self.server_jar, str)
            or not self.server_jar.startswith("local:///")
            or not self.server_jar.endswith(".jar")
            or any(c.isspace() for c in self.server_jar)
        ):
            raise ValueError("server_jar must be an explicit local:/// path to a JAR in the image")
        _name(self.namespace, "namespace")
        _name(self.service_account, "service_account", max_length=253)
        if not isinstance(self.app_name, str) or not re.fullmatch(
            r"[a-z](?:[a-z0-9-]*[a-z0-9])?", self.app_name
        ):
            raise ValueError("app_name must start with a lowercase letter and contain DNS-label characters")
        _port(self.port)
        if self.image_pull_policy not in {"Always", "IfNotPresent", "Never"}:
            raise ValueError("image_pull_policy must be Always, IfNotPresent, or Never")
        for name, resources in (("driver", self.driver_resources), ("executor", self.executor_resources)):
            if not isinstance(resources, PodResources):
                raise ValueError(f"{name}_resources must be PodResources")
            for key in ("cpu", "memory", "memory_overhead"):
                _positive(getattr(resources, key), f"{name}.{key}", zero=key == "memory_overhead")
        instances = self.executor_instances
        if not isinstance(instances, ExecutorInstances):
            raise ValueError("executor_instances must be ExecutorInstances")
        for key in ("min", "max", "initial"):
            value = getattr(instances, key)
            if value is not None:
                _positive(value, f"executor_instances.{key}", zero=key != "max")
        if instances.min is not None and instances.max is not None and instances.min > instances.max:
            raise ValueError("Minimum executors cannot exceed maximum executors")
        if instances.initial is not None and (
            (instances.min is not None and instances.initial < instances.min)
            or (instances.max is not None and instances.initial > instances.max)
        ):
            raise ValueError("Initial executors must be within the configured minimum and maximum")
        if (
            type(instances.executor_allocation_ratio) not in {int, float}
            or not math.isfinite(instances.executor_allocation_ratio)
            or not 0 < instances.executor_allocation_ratio <= 1
        ):
            raise ValueError("Executor allocation ratio must be finite and in (0, 1]")
        for value in (instances.scheduler_backlog_timeout, instances.sustained_scheduler_backlog_timeout):
            if value is not None and (
                not isinstance(value, str) or not re.fullmatch(r"[1-9][0-9]*(?:ms|s|m|h|d)?", value)
            ):
                raise ValueError("Executor backlog timeouts must be positive Spark durations")
        if not isinstance(self.spark_conf, dict) or any(
            not isinstance(k, str) or not k or not isinstance(v, str) for k, v in self.spark_conf.items()
        ):
            raise ValueError("spark_conf must be a string-to-string mapping")
        for name in ("driver_node_selector", "executor_node_selector"):
            _node_selector(getattr(self, name), name)
        for name in ("driver_tolerations", "executor_tolerations"):
            _tolerations(getattr(self, name), name)
        if not isinstance(self.image_pull_secrets, list):
            raise ValueError("image_pull_secrets must be a list of Secret names")
        for secret in self.image_pull_secrets:
            _name(secret, "image_pull_secrets entry", max_length=253)
        if len(set(self.image_pull_secrets)) != len(self.image_pull_secrets):
            raise ValueError("image_pull_secrets must not repeat a Secret")
        for name in ("driver_env_from_secrets", "executor_env_from_secrets"):
            _secret_environment(getattr(self, name), name)
        if any(f"spark.executorEnv.{name}" in self.spark_conf for name in self.executor_env_from_secrets):
            raise ValueError("executor_env_from_secrets and spark_conf cannot both set an executor variable")
        if any(
            (key in _PROTECTED_CONF or key.startswith(_PROTECTED_PREFIXES))
            and key != "spark.kubernetes.executor.request.cores"
            for key in self.spark_conf
        ):
            raise ValueError(
                "spark_conf cannot override managed identity, ports, images, resources, or allocation"
            )
        request_cores = self.spark_conf.get("spark.kubernetes.executor.request.cores")
        if request_cores is not None:
            if not re.fullmatch(r"(?:[0-9]+(?:\.[0-9]+)?|\.[0-9]+)m?", request_cores):
                raise ValueError("Executor CPU request must be a positive Kubernetes CPU quantity")
            amount = float(request_cores.removesuffix("m")) / (1000 if request_cores.endswith("m") else 1)
            if not math.isfinite(amount) or amount <= 0 or amount > self.executor_resources.cpu:
                raise ValueError(
                    "Executor CPU request must be positive and no greater than configured executor cores"
                )
        if not isinstance(self.packages, list) or any(
            not isinstance(p, str) or not re.fullmatch(r"[^\s,:]+:[^\s,:]+:[^\s,:]+", p)
            for p in self.packages
        ):
            raise ValueError("packages must contain explicit Maven group:artifact:version coordinates")
        validate_max_runtime(self.max_runtime)

ConnectServerCreateError

Bases: ConnectServerError

Creation failed; inspect run, phase and cleanup for partial resource effects.

Source code in spark_on_k8s/connect_server.py

class ConnectServerCreateError(ConnectServerError):
    """Creation failed; inspect run, phase and cleanup for partial resource effects."""

ConnectServerError

Bases: RuntimeError

Server operation failed without interpreting uncertainty as absence.

Source code in spark_on_k8s/connect_server.py

class ConnectServerError(RuntimeError):
    """Server operation failed without interpreting uncertainty as absence."""

    def __init__(self, message, *, handle=None, run=None, phase=None, cleanup=None):
        super().__init__(message)
        self.handle = handle
        self.run = handle.run if handle is not None else run
        self.phase = phase
        self.cleanup = cleanup or {"status": "not_attempted", "reason": "No rollback was requested"}

ConnectServerHandle dataclass

Credential-free identity for a server and both owned Kubernetes Services.

Source code in spark_on_k8s/connect_server.py

@dataclass(frozen=True, kw_only=True)
class ConnectServerHandle:
    """Credential-free identity for a server and both owned Kubernetes Services."""

    run: RunHandle
    service_name: str
    service_uid: str
    driver_service_uid: str
    port: int

    def __post_init__(self):
        if not isinstance(self.run, RunHandle):
            raise ValueError("run must be a RunHandle")
        _name(self.run.namespace, "namespace")
        _name(self.run.app_id, "app_id")
        _name(self.run.driver_pod, "driver_pod", max_length=253)
        _name(self.service_name, "service_name")
        if (
            self.run.driver_pod != f"{self.run.app_id}-driver"
            or self.service_name != f"{self.run.app_id}-connect"
        ):
            raise ValueError("Connect resource names must match the recorded application identity")
        for value in (self.run.driver_uid, self.service_uid, self.driver_service_uid):
            _uid(value)
        _port(self.port)

    @property
    def endpoint(self) -> str:
        """Return the private cluster DNS endpoint; no credentials are included."""
        return f"sc://{self.service_name}.{self.run.namespace}.svc:{self.port}"

    def to_dict(self) -> dict[str, Any]:
        return {
            "run": self.run.to_dict(),
            "service_name": self.service_name,
            "service_uid": self.service_uid,
            "driver_service_uid": self.driver_service_uid,
            "port": self.port,
        }

    @classmethod
    def from_dict(cls, value: dict) -> ConnectServerHandle:
        if not isinstance(value, dict) or set(value) != {
            "run",
            "service_name",
            "service_uid",
            "driver_service_uid",
            "port",
        }:
            raise ValueError("Invalid Connect server receipt fields")
        run = value["run"]
        if not isinstance(run, dict) or set(run) != {"namespace", "app_id", "driver_pod", "driver_uid"}:
            raise ValueError("Connect server receipt requires complete run identity")
        return cls(**{**value, "run": RunHandle(**run)})

endpoint property

Return the private cluster DNS endpoint; no credentials are included.

ConnectServerOwnershipError

Bases: ConnectServerError

Observed resources do not match the recorded identity; no adoption is allowed.

Source code in spark_on_k8s/connect_server.py

class ConnectServerOwnershipError(ConnectServerError):
    """Observed resources do not match the recorded identity; no adoption is allowed."""

ConnectServerTimeout

Bases: TimeoutError

Readiness observation expired; the server was not deleted.

Source code in spark_on_k8s/connect_server.py

class ConnectServerTimeout(TimeoutError):
    """Readiness observation expired; the server was not deleted."""

    def __init__(self, handle, last_failure: str | None = None):
        message = "Spark Connect readiness timed out; the server was not stopped or deleted"
        super().__init__(message + (f". Last query check: {last_failure}" if last_failure else ""))
        self.handle = handle
        self.run = handle.run
        self.phase = "readiness"
        self.cleanup = {"status": "not_attempted", "reason": "Readiness timeout does not delete the server"}

connect_server_status(client, handle, *, timeout=10)

Read matching resource state. Only confirmed 404 means missing; errors propagate.

Source code in spark_on_k8s/connect_server.py

def connect_server_status(client: SparkOnK8S, handle: ConnectServerHandle, *, timeout: float = 10) -> dict:
    """Read matching resource state. Only confirmed 404 means missing; errors propagate."""
    if not isinstance(handle, ConnectServerHandle):
        raise ValueError("handle must be ConnectServerHandle")
    deadline = _deadline(timeout)
    with client.k8s_client_manager.client() as raw:
        pod, transport, service = _resources(_api(raw), handle, deadline)
    missing = [
        name
        for name, obj in (("driver", pod), ("driver_service", transport), ("connect_service", service))
        if obj is None
    ]
    phase = pod.status.phase if pod is not None and pod.status else None
    # Kubernetes sets a pod-level reason such as DeadlineExceeded (max_runtime) or Evicted.
    reason = getattr(pod.status, "reason", None) if pod is not None and pod.status else None
    deleting = any(
        obj is not None and obj.metadata.deletion_timestamp is not None for obj in (pod, transport, service)
    )
    created, age = _created(pod, datetime.now(UTC))
    return {
        "status": "missing" if missing else _driver_state(phase, deleting),
        "driver_phase": phase,
        "driver_reason": reason if isinstance(reason, str) else None,
        "missing": missing,
        "endpoint": handle.endpoint,
        "created_at": _timestamp(created),
        "age_seconds": age,
        # Kubernetes observation cannot learn the server version; only an optional label is reported.
        "spark_version": (pod.metadata.labels or {}).get("spark-version") if pod is not None else None,
        "query_ready": None,
        "message": "Kubernetes observation only; query readiness has not been checked",
    }

create_connect_server(client, config)

Create a driver and private gRPC Service, without claiming query readiness.

A failed create attempts rollback only after receiving a confirmed driver UID. A timeout with no confirmed UID is uncertain: no retry, adoption or deletion occurs. The raised error retains the planned/confirmed run and rollback outcome.

Source code in spark_on_k8s/connect_server.py

def create_connect_server(client: SparkOnK8S, config: ConnectServerConfig) -> ConnectServerHandle:
    """Create a driver and private gRPC Service, without claiming query readiness.

    A failed create attempts rollback only after receiving a confirmed driver UID.
    A timeout with no confirmed UID is uncertain: no retry, adoption or deletion occurs.
    The raised error retains the planned/confirmed run and rollback outcome.
    """
    if not isinstance(config, ConnectServerConfig):
        raise ValueError("config must be ConnectServerConfig")
    config = deepcopy(config)
    config.__post_init__()  # Nested resource/mapping objects may have changed since construction.
    token = uuid4().hex
    options = _options(config, token)
    app_id = f"{options['app_name']}-{token}"
    planned = RunHandle(namespace=config.namespace, app_id=app_id, driver_pod=f"{app_id}-driver")
    run = None
    handle = None
    phase = "driver_submission"
    try:
        run = client.submit_run(**options)
        _uid(run.driver_uid)
        phase = "driver_service_verification"
        with client.k8s_client_manager.client() as raw:
            api = _api(raw)
            deadline = WaitDeadline(10)
            pod = _read(api, "pod", run.driver_pod, run.namespace, deadline)
            transport = _read(api, "service", run.app_id, run.namespace, deadline)
            _driver(run, pod)
            _service(run, transport, name=run.app_id, uid=None)
            if pod is None or transport is None:
                raise ConnectServerError("Submitted Connect driver or driver Service is missing", run=run)
            phase = "grpc_service_creation"
            name = f"{run.app_id}-connect"
            body = _grpc_service(
                run.app_id,
                run.namespace,
                config.port,
                owner=k8s.V1OwnerReference(
                    api_version="v1", kind="Pod", name=run.driver_pod, uid=run.driver_uid
                ),
            )
            service = api.create_namespaced_service(
                namespace=run.namespace, body=body, **deadline.request_options()
            )
            handle = ConnectServerHandle(
                run=run,
                service_name=name,
                service_uid=service.metadata.uid,
                driver_service_uid=transport.metadata.uid,
                port=config.port,
            )
            _service(run, service, name=name, uid=handle.service_uid, port=config.port)
        return handle
    except Exception as exc:
        confirmed = run or client.last_submitted_run
        if confirmed is not None and (
            confirmed.namespace != planned.namespace
            or confirmed.app_id != planned.app_id
            or confirmed.driver_pod != planned.driver_pod
        ):
            confirmed = None
        cleanup = {"status": "not_attempted", "reason": "Driver creation is unconfirmed; resources may exist"}
        if confirmed is not None and confirmed.driver_uid:
            try:
                client.cancel_run(confirmed)
                cleanup = {
                    "status": "requested",
                    "reason": (
                        "UID-checked driver deletion requested; owned resources "
                        "are garbage-collected asynchronously"
                    ),
                }
            except KeyboardInterrupt as interrupted:
                interrupted.handle = handle
                interrupted.run = confirmed
                interrupted.phase = "rollback"
                interrupted.cleanup = {
                    "status": "unknown",
                    "reason": "Rollback was interrupted; deletion may already have been requested",
                }
                raise
            except Exception:
                cleanup = {
                    "status": "failed",
                    "reason": "UID-checked rollback failed; inspect the recorded run before retrying cleanup",
                }
        raise ConnectServerCreateError(
            "Spark Connect server creation failed",
            handle=handle,
            run=confirmed or planned,
            phase=phase,
            cleanup=cleanup,
        ) from exc

delete_connect_server(client, handle, *, timeout=10)

Verify all three resource identities before requesting UID-conditional deletion.

Missing resources are idempotent. Success means deletion was requested, not that executors or Services have already disappeared. A concurrent change can interrupt a partially applied deletion; UID preconditions protect replacement resources.

Source code in spark_on_k8s/connect_server.py

def delete_connect_server(client: SparkOnK8S, handle: ConnectServerHandle, *, timeout: float = 10) -> dict:
    """Verify all three resource identities before requesting UID-conditional deletion.

    Missing resources are idempotent. Success means deletion was requested, not that
    executors or Services have already disappeared. A concurrent change can interrupt
    a partially applied deletion; UID preconditions protect replacement resources.
    """
    if not isinstance(handle, ConnectServerHandle):
        raise ValueError("handle must be ConnectServerHandle")
    deadline = _deadline(timeout)
    requested = []
    with client.k8s_client_manager.client() as raw:
        api = _api(raw)
        pod, transport, service = _resources(api, handle, deadline)
        for kind, obj in (("service", service), ("service", transport), ("pod", pod)):
            if obj is None:
                continue
            try:
                getattr(api, f"delete_namespaced_{kind}")(
                    name=obj.metadata.name,
                    namespace=handle.run.namespace,
                    body=k8s.V1DeleteOptions(
                        propagation_policy="Foreground",
                        preconditions=k8s.V1Preconditions(uid=obj.metadata.uid),
                    ),
                    **deadline.request_options(),
                )
            except ApiException as exc:
                if exc.status != 404:
                    raise
            requested.append({"kind": kind, "name": obj.metadata.name, "uid": obj.metadata.uid})
    return {
        "status": "deletion_requested" if requested else "absent",
        "requested": requested,
        "message": "Resource deletion and executor garbage collection are asynchronous",
    }

list_connect_servers(client, namespace, *, timeout=10)

List driver pods labeled as servers created by this helper in one namespace, oldest first.

This is a read-only, label-based observation: it neither verifies receipts, Services or query readiness nor deletes anything. A matching label is not proof of ownership; deletion still requires the recorded identity checks. Errors propagate.

Source code in spark_on_k8s/connect_server.py

def list_connect_servers(client: SparkOnK8S, namespace: str, *, timeout: float = 10) -> list[dict]:
    """List driver pods labeled as servers created by this helper in one namespace, oldest first.

    This is a read-only, label-based observation: it neither verifies receipts, Services
    or query readiness nor deletes anything. A matching label is not proof of ownership;
    deletion still requires the recorded identity checks. Errors propagate.
    """
    _name(namespace, "namespace")
    deadline = _deadline(timeout)
    with client.k8s_client_manager.client() as raw:
        pods = _api(raw).list_namespaced_pod(
            namespace=namespace,
            label_selector=f"spark-role=driver,{_SERVER_LABEL}=true",
            **deadline.request_options(),
        )
    now = datetime.now(UTC)
    servers = []
    for pod in pods.items:
        labels = pod.metadata.labels or {}
        if not labels.get("spark-app-id"):
            continue  # The helper always records an application ID on its driver.
        phase = pod.status.phase if pod.status else None
        containers = (pod.spec.containers if pod.spec else None) or []
        name = driver_container_name(pod) if containers else None
        created, age = _created(pod, now)
        servers.append(
            (
                created,
                {
                    "app_id": labels["spark-app-id"],
                    "name": labels.get("spark-app-name"),
                    "namespace": pod.metadata.namespace or namespace,
                    "driver_pod": pod.metadata.name,
                    "driver_uid": pod.metadata.uid,
                    "status": _driver_state(phase, pod.metadata.deletion_timestamp is not None),
                    "driver_phase": phase,
                    "spark_version": labels.get("spark-version"),
                    "image": next((c.image for c in containers if c.name == name), None),
                    "created_at": _timestamp(created),
                    "age_seconds": age,
                },
            )
        )
    servers.sort(key=lambda item: (item[0] is None, item[0] or now, item[1]["app_id"]))
    return [server for _, server in servers]

plan_connect_server(client, config)

Render the resources create_connect_server would request, without cluster access.

The plan uses the same submission builder as creation plus the gRPC Service. Secret settings appear only as Secret names and keys: no Secret is read, so no value can appear. Identifiers are illustrative; nothing is created or reserved.

Source code in spark_on_k8s/connect_server.py

def plan_connect_server(client: SparkOnK8S, config: ConnectServerConfig) -> ApplicationPlan:
    """Render the resources create_connect_server would request, without cluster access.

    The plan uses the same submission builder as creation plus the gRPC Service. Secret
    settings appear only as Secret names and keys: no Secret is read, so no value can
    appear. Identifiers are illustrative; nothing is created or reserved.
    """
    if not isinstance(config, ConnectServerConfig):
        raise ValueError("config must be ConnectServerConfig")
    config = deepcopy(config)
    config.__post_init__()
    plan = client.plan_app(**_options(config, "preview"))
    service = _grpc_service(plan.app_id, plan.namespace, config.port)
    with k8s.ApiClient() as serializer:
        service = {**serializer.sanitize_for_serialization(service), "apiVersion": "v1", "kind": "Service"}
    return replace(plan, resources=(*plan.resources, service))

wait_connect_server_ready(client, handle, *, profile, timeout=60, poll_interval=1)

Require a real query against this application, then revalidate Kubernetes identity.

Source code in spark_on_k8s/connect_server.py

def wait_connect_server_ready(
    client: SparkOnK8S,
    handle: ConnectServerHandle,
    *,
    profile: ConnectProfile,
    timeout: float = 60,
    poll_interval: float = 1,
) -> dict:
    """Require a real query against this application, then revalidate Kubernetes identity."""
    if not isinstance(profile, ConnectProfile):
        raise ValueError("profile must be a reachable ConnectProfile")
    if type(poll_interval) not in {int, float} or not math.isfinite(poll_interval) or poll_interval <= 0:
        raise ValueError("poll_interval must be positive and finite")
    deadline = _deadline(timeout)
    last_failure = None
    try:
        while True:
            status = connect_server_status(client, handle, timeout=deadline.remaining())
            if status["status"] in {"missing", "deleting", "terminated"}:
                raise ConnectServerError(
                    "Connect server is missing, terminating, or terminated", handle=handle, phase="readiness"
                )
            if status["status"] == "running":
                try:
                    # The same per-check budget as `connect check`: a slower client start is not unreadiness.
                    result = profile.probe(
                        timeout=min(15, deadline.remaining()), expected_app_id=handle.run.app_id
                    )
                except ConnectClientUnavailable:
                    # A missing local client is not a server state; polling cannot make it ready.
                    raise
                except (ConnectError, TimeoutError) as failure:
                    # Reported if the wait times out, for example when another application answers.
                    last_failure = str(failure)
                else:
                    after = connect_server_status(client, handle, timeout=deadline.remaining())
                    if after["status"] == "running":
                        return {
                            **after,
                            "status": "ready",
                            "query_ready": True,
                            # The query reports the running server's version, unlike the optional label.
                            "spark_version": result.get("spark_version"),
                            "client_version": result.get("client_version"),
                            "warnings": list(result.get("warnings") or []),
                            "probe": result,
                            "message": "Connect query and application identity verified",
                        }
            deadline.sleep(poll_interval)
    except TimeoutError:
        raise ConnectServerTimeout(handle, last_failure) from None
    except (TransportTimeoutError, MaxRetryError) as error:
        # urllib3 can wrap a request timeout once the remaining wait budget is tiny.
        # Only the enclosing deadline expiring changes that transport error into a wait timeout.
        reason = error.reason if isinstance(error, MaxRetryError) else error
        if isinstance(reason, TransportTimeoutError) and not isinstance(reason, NewConnectionError):
            try:
                deadline.remaining()
            except TimeoutError:
                raise ConnectServerTimeout(handle, last_failure) from None
        raise