spark_on_k8s.spark_operator_migration

Implementation reference. These symbols describe how the project works; their presence here does not make them supported application imports. Use the documented entry points and support policy.

Read-only assessment of Kubeflow Spark Operator manifests for a SparkJob migration.

The assessor reads sparkoperator.k8s.io/v1beta2 SparkApplication and ScheduledSparkApplication documents and gives every field one accounting status: mapped, approximated, unsupported or ignored. It drafts a SparkJob document from the settings that translate and validates that draft offline. It never contacts a cluster and never translates a field it does not recognize. Values that look like credentials are neither shown nor copied into a draft.

Finding dataclass

The accounting status of one source field; it covers any fields nested below it.

Source code in spark_on_k8s/spark_operator_migration.py

@dataclass(frozen=True)
class Finding:
    """The accounting status of one source field; it covers any fields nested below it."""

    path: str
    status: str
    target: str | None = None
    note: str | None = None
    alternative: str | None = None

    def to_dict(self) -> dict[str, Any]:
        return {
            "path": self.path,
            "status": self.status,
            "target": self.target,
            "note": self.note,
            "alternative": self.alternative,
        }

ManifestError

Bases: ValueError

The input cannot be read as Spark Operator manifests.

Source code in spark_on_k8s/spark_operator_migration.py

class ManifestError(ValueError):
    """The input cannot be read as Spark Operator manifests."""

assess_document(document, *, position=1)

Assess one SparkApplication or ScheduledSparkApplication mapping without contacting a cluster.

Source code in spark_on_k8s/spark_operator_migration.py

def assess_document(document: dict, *, position: int = 1) -> dict[str, Any]:
    """Assess one SparkApplication or ScheduledSparkApplication mapping without contacting a cluster."""
    kind = document.get("kind")
    metadata = document.get("metadata")
    if not isinstance(metadata, dict) or not _text(metadata.get("name")):
        raise ManifestError(f"Document {position}: metadata.name is required.")
    spec = document.get("spec")
    if not isinstance(spec, dict):
        raise ManifestError(f"Document {position}: spec must be an object.")
    application = spec.get("template") if kind == "ScheduledSparkApplication" else spec
    if not isinstance(application, dict):
        raise ManifestError(f"Document {position}: spec.template must be an object.")
    a = _Assessment()
    job_name = _job_name(metadata["name"])
    prefix = "spec.template" if kind == "ScheduledSparkApplication" else "spec"
    if kind == "ScheduledSparkApplication":
        _scheduled(a, spec)
    _application(a, application, prefix)
    _metadata(a, metadata, job_name)
    for key in document:
        if key in {"apiVersion", "kind"}:
            a.record(key, MAPPED, key, f"The draft is a {API_VERSION} SparkJob document.")
        elif key == "status":
            a.record(key, IGNORED, note="Observed state written by the operator.")
        elif key not in {"metadata", "spec"}:
            a.unknown(key)
    _account_for_every_leaf(a, document)
    draft = {"apiVersion": API_VERSION, "kind": "SparkJob", "metadata": {"name": job_name}}
    draft["spec"] = _draft_spec(a)
    draft_error = None
    try:
        JobSpec.from_mapping(copy.deepcopy(draft)).plan()
    except (ValueError, TypeError, KeyError, AttributeError) as error:
        draft_error = str(error) or type(error).__name__
    findings = sorted(a.findings, key=lambda finding: finding.path)
    counts = {status: sum(finding.status == status for finding in findings) for status in STATUSES}
    namespace = metadata.get("namespace")
    return {
        "document": position,
        "kind": kind,
        "name": metadata["name"],
        "namespace": namespace if isinstance(namespace, str) else None,
        "migration_ready": draft_error is None and counts[UNSUPPORTED] == 0,
        "counts": counts,
        "findings": [finding.to_dict() for finding in findings],
        "notes": _notes(a, kind),
        "draft": draft if draft_error is None else None,
        "draft_error": draft_error,
        "draft_file": None,
    }

assess_file(path)

Read and assess a manifest file. Nothing is submitted and no cluster is contacted.

Source code in spark_on_k8s/spark_operator_migration.py

def assess_file(path: str | Path) -> dict[str, Any]:
    """Read and assess a manifest file. Nothing is submitted and no cluster is contacted."""
    with Path(path).expanduser().open("rb") as stream:
        return assess_manifests(stream.read(MAX_JOB_BYTES + 1))

assess_manifests(content)

Assess every Spark Operator application in a YAML stream; other documents are skipped.

Source code in spark_on_k8s/spark_operator_migration.py

def assess_manifests(content: bytes | str) -> dict[str, Any]:
    """Assess every Spark Operator application in a YAML stream; other documents are skipped."""
    applications, skipped = [], []
    for position, document in enumerate(load_manifests(content), start=1):
        if document is None:
            continue
        if not isinstance(document, dict):
            raise ManifestError(f"Document {position} is not a mapping.")
        api_version, kind = document.get("apiVersion"), document.get("kind")
        if api_version == OPERATOR_API_VERSION and kind in OPERATOR_KINDS:
            applications.append(assess_document(document, position=position))
        else:
            skipped.append(
                {
                    "document": position,
                    "api_version": api_version if isinstance(api_version, str) else None,
                    "kind": kind if isinstance(kind, str) else None,
                    "reason": f"Not an {OPERATOR_API_VERSION} SparkApplication or ScheduledSparkApplication.",
                }
            )
    if not applications:
        raise ManifestError(
            f"No {OPERATOR_API_VERSION} SparkApplication or ScheduledSparkApplication document was found."
        )
    return {
        "schema_version": SCHEMA_VERSION,
        "operation": "assess_operator",
        "error": None,
        "migration_ready": all(item["migration_ready"] for item in applications),
        "applications": applications,
        "skipped": skipped,
    }

draft_yaml(application, source)

Render a validated draft with a header that marks it as a migration draft.

Source code in spark_on_k8s/spark_operator_migration.py

def draft_yaml(application: dict[str, Any], source: str) -> str:
    """Render a validated draft with a header that marks it as a migration draft."""
    location = f"{application['namespace']}/" if application["namespace"] else ""
    header = (
        f"# DRAFT converted by `spark-on-k8s job assess-operator` from {application['kind']}\n"
        f"# {location}{application['name']} in {source}.\n"
        "# It contains only the settings reported as mapped or approximated. Review every\n"
        "# approximated and unsupported finding, then validate before use:\n"
        "#   spark-on-k8s job validate <this file>\n"
    )
    return header + yaml.safe_dump(application["draft"], sort_keys=False)

load_manifests(content)

Parse a YAML stream of at most 1 MiB, rejecting aliases and duplicate keys.

Source code in spark_on_k8s/spark_operator_migration.py

def load_manifests(content: bytes | str) -> list[Any]:
    """Parse a YAML stream of at most 1 MiB, rejecting aliases and duplicate keys."""
    if len(content) > MAX_JOB_BYTES:
        raise ManifestError("Manifests must be no larger than 1 MiB.")
    try:
        return list(yaml.load_all(content, Loader=_ManifestLoader))
    except yaml.YAMLError as error:
        mark = getattr(error, "problem_mark", None)
        where = f" at line {mark.line + 1}, column {mark.column + 1}" if mark is not None else ""
        # Do not echo YAML snippets: the input can contain credentials.
        raise ManifestError(f"Invalid YAML{where}.") from error