spark_on_k8s.jobs

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

JobSpec dataclass

Validated effective job settings, independent of Airflow and the HTTP API.

Source code in spark_on_k8s/jobs.py

@dataclass(frozen=True)
class JobSpec:
    """Validated effective job settings, independent of Airflow and the HTTP API."""

    name: str
    spec: dict[str, Any]
    kubernetes: dict[str, Any]
    profile: str | None = None

    @classmethod
    def from_mapping(cls, document: dict, *, profile: str | None = None) -> JobSpec:
        _object(document, {"apiVersion", "kind", "metadata", "spec", "kubernetes", "profiles"}, "job")
        if document.get("apiVersion") != API_VERSION or document.get("kind") != "SparkJob":
            raise JobValidationError(f"Expected apiVersion: {API_VERSION} and kind: SparkJob.")
        metadata = _object(document.get("metadata", {}), {"name"}, "metadata")
        name = metadata.get("name")
        _text(name, "metadata.name")
        if len(name) > 63 or not _DNS_NAME.fullmatch(name) or not name[0].isalpha():
            raise JobValidationError(
                "metadata.name must start with a letter and be a DNS label of at most 63 characters."
            )
        profiles = _object(document.get("profiles", {}), None, "profiles")
        for overlay in profiles.values():
            _object(overlay, {"spec", "kubernetes"}, "profile")
        resolved = copy.deepcopy(document)
        if profile is not None:
            if profile not in profiles:
                raise JobValidationError(f"Unknown profile: {profile}.")
            resolved = _merge(resolved, profiles[profile])
        spec = _object(resolved.get("spec", {}), _FIELDS, "spec")
        _validate_spec(spec)
        connection = _object(
            resolved.get("kubernetes", {}), {"context", "config_file", "in_cluster"}, "kubernetes"
        )
        for field in {"context", "config_file"} & connection.keys():
            _text(connection[field], f"kubernetes.{field}")
        if "in_cluster" in connection and type(connection["in_cluster"]) is not bool:
            raise JobValidationError("kubernetes.in_cluster must be a boolean.")
        if connection.get("in_cluster") and ("context" in connection or "config_file" in connection):
            raise JobValidationError(
                "In-cluster authentication cannot be combined with a kubeconfig/context."
            )
        return cls(name, copy.deepcopy(spec), copy.deepcopy(connection), profile)

    @classmethod
    def from_file(cls, path: str | Path, *, profile: str | None = None) -> JobSpec:
        path, document = cls._read_file(path)
        result = cls.from_mapping(document, profile=profile)
        for field in ("driver_pod_template", "executor_pod_template"):
            value = result.spec.get(field)
            if isinstance(value, str) and "\n" not in value and value.endswith((".yaml", ".yml")):
                template_path = Path(value).expanduser()
                if not template_path.is_absolute():
                    template_path = path.parent / template_path
                result.spec[field] = template_path.read_text(encoding="utf-8")
        if "config_file" in result.kubernetes:
            config_path = Path(result.kubernetes["config_file"]).expanduser()
            if not config_path.is_absolute():
                config_path = path.parent / config_path
            result.kubernetes["config_file"] = str(config_path)
        return result

    @staticmethod
    def _read_file(path):
        path = Path(path).expanduser().resolve()
        with path.open("rb") as stream:
            content = stream.read(MAX_JOB_BYTES + 1)
        if len(content) > MAX_JOB_BYTES:
            raise JobValidationError("Job files must be no larger than 1 MiB.")
        try:
            document = yaml.load(content, Loader=_UniqueLoader)
        except JobValidationError:
            raise
        except yaml.YAMLError as error:
            raise JobValidationError("Invalid job YAML.") from error
        except (ValueError, KeyError, AttributeError) as error:
            # PyYAML builds resolved scalars such as 2024-02-30 or 0x_ with plain, non-YAML errors.
            raise JobValidationError(
                "Invalid job YAML: a value is not a valid number, boolean or timestamp; quote text values."
            ) from error
        except RecursionError as error:
            raise JobValidationError("Invalid job YAML: the document is nested too deeply.") from error
        return path, document

    @classmethod
    def validate_profiles(cls, path: str | Path) -> list[dict[str, Any]]:
        """Validate the base and every named profile offline, reporting every failure."""
        _, document = cls._read_file(path)
        _object(document, None, "job")
        profiles = _object(document.get("profiles", {}), None, "profiles")
        results = []
        for profile in [None, *sorted(profiles)]:
            try:
                cls.from_file(path, profile=profile).plan()
            except (ValueError, TypeError, OSError) as error:
                results.append({"profile": profile, "valid": False, "error": validation_error(error)})
            else:
                results.append({"profile": profile, "valid": True, "error": None})
        return results

    def to_submit_kwargs(self) -> dict[str, Any]:
        _validate_spec(self.spec)
        # Job files are portable: ambient CLI defaults must not change their meaning.
        options = {
            "namespace": "default",
            "service_account": "spark",
            "spark_conf": {},
            "class_name": None,
            "packages": [],
            "app_arguments": [],
            "app_waiter": "no_wait",
            "image_pull_policy": "IfNotPresent",
            "ui_reverse_proxy": False,
            "driver_resources": {},
            "executor_resources": {},
            "executor_instances": {"initial": 2},
            "executor_pod_template": "",
            "executor_pod_template_path": "",
            "secret_values": {},
            "driver_env_vars_from_secrets": [],
            "startup_timeout": 0,
            "driver_node_selector": {},
            "executor_node_selector": {},
            "driver_labels": {},
            "executor_labels": {},
            "driver_annotations": {},
            "executor_annotations": {},
            **copy.deepcopy(self.spec),
        }
        options["app_name"] = self.name
        for field in _RESOURCE_FIELDS & options.keys():
            options[field] = PodResources(**options[field])
        if "executor_instances" in options:
            instances = options["executor_instances"]
            instances.setdefault(
                "initial", instances.get("min", 0) if {"min", "max"} & instances.keys() else 2
            )
            options["executor_instances"] = ExecutorInstances(**instances)
        if isinstance(options.get("executor_pod_template"), dict):
            options["executor_pod_template"] = yaml.safe_dump(options["executor_pod_template"])
        return options

    def to_operator_kwargs(self) -> dict[str, Any]:
        """Use the same settings in Airflow; choose Kubernetes credentials through an Airflow connection."""
        if self.kubernetes:
            raise JobValidationError(
                "Use kubernetes_conn_id for Airflow credentials "
                "and a job profile without kubernetes settings."
            )
        return self.to_submit_kwargs()

    def client(self) -> SparkOnK8S:
        return SparkOnK8S(
            k8s_client_manager=KubernetesClientManager(
                **{
                    "context": None,
                    "config_file": None,
                    "in_cluster": False,
                    "client_configuration": None,
                    **self.kubernetes,
                }
            )
        )

    def plan(self) -> ApplicationPlan:
        return self.client().plan_app(**self.to_submit_kwargs())

    def submit(self) -> str:
        return self.client().submit_app(**self.to_submit_kwargs())

to_operator_kwargs()

Use the same settings in Airflow; choose Kubernetes credentials through an Airflow connection.

Source code in spark_on_k8s/jobs.py

def to_operator_kwargs(self) -> dict[str, Any]:
    """Use the same settings in Airflow; choose Kubernetes credentials through an Airflow connection."""
    if self.kubernetes:
        raise JobValidationError(
            "Use kubernetes_conn_id for Airflow credentials "
            "and a job profile without kubernetes settings."
        )
    return self.to_submit_kwargs()

validate_profiles(path) classmethod

Validate the base and every named profile offline, reporting every failure.

Source code in spark_on_k8s/jobs.py

@classmethod
def validate_profiles(cls, path: str | Path) -> list[dict[str, Any]]:
    """Validate the base and every named profile offline, reporting every failure."""
    _, document = cls._read_file(path)
    _object(document, None, "job")
    profiles = _object(document.get("profiles", {}), None, "profiles")
    results = []
    for profile in [None, *sorted(profiles)]:
        try:
            cls.from_file(path, profile=profile).plan()
        except (ValueError, TypeError, OSError) as error:
            results.append({"profile": profile, "valid": False, "error": validation_error(error)})
        else:
            results.append({"profile": profile, "valid": True, "error": None})
    return results

JobValidationError

Bases: ValueError

A job definition has an invalid structure or unsupported setting.

Source code in spark_on_k8s/jobs.py

class JobValidationError(ValueError):
    """A job definition has an invalid structure or unsupported setting."""

validation_error(error)

Classify an offline validation failure for the versioned validate result.

Source code in spark_on_k8s/jobs.py

def validation_error(error: Exception) -> dict[str, str]:
    """Classify an offline validation failure for the versioned validate result."""
    # An unreadable job or template file is a read failure, not an invalid definition.
    return {"code": "read_error" if isinstance(error, OSError) else "invalid_job", "message": str(error)}