Skip to content

spark_on_k8s.cli.job

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.

assess_operator(file, output, draft_dir)

Assess Kubeflow Spark Operator manifests offline and draft SparkJob files.

Reads sparkoperator.k8s.io/v1beta2 SparkApplication and ScheduledSparkApplication documents without contacting a cluster. Unsupported fields are reported, not translated, and do not change the exit code; unreadable input exits 1.

Source code in spark_on_k8s/cli/job.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
@job_cli.command(name="assess-operator")
@click.argument("file", type=click.Path(exists=True, dir_okay=False))
@click.option("--output", type=click.Choice(["text", "json"]), default="text", show_default=True)
@click.option(
    "--draft-dir",
    type=click.Path(file_okay=False),
    help="Write each valid SparkJob draft to this directory, created if missing, as NAME.yaml; "
    "existing files are kept.",
)
def assess_operator(file, output, draft_dir):
    """Assess Kubeflow Spark Operator manifests offline and draft SparkJob files.

    Reads sparkoperator.k8s.io/v1beta2 SparkApplication and ScheduledSparkApplication
    documents without contacting a cluster. Unsupported fields are reported, not
    translated, and do not change the exit code; unreadable input exits 1.
    """
    from spark_on_k8s.spark_operator_migration import SCHEMA_VERSION, ManifestError, assess_file

    result = {
        "schema_version": SCHEMA_VERSION,
        "operation": "assess_operator",
        "error": None,
        "migration_ready": False,
        "applications": None,
        "skipped": None,
    }
    try:
        result.update(assess_file(file))
        if draft_dir is not None:
            Path(draft_dir).mkdir(parents=True, exist_ok=True)
            _write_drafts(result["applications"], Path(draft_dir), Path(file).name)
    except (ManifestError, OSError) as error:
        result["error"] = {"code": "operation_error", "message": str(error)}
    if output == "json":
        click.echo(json.dumps(result, ensure_ascii=False, default=str))
    else:
        _echo_assessment(result)
    if result["error"]:
        raise click.exceptions.Exit(1)

plan(file, profile, output, max_cpu, max_memory_mib)

Print a redacted offline preview. Nothing is submitted or reserved.

Source code in spark_on_k8s/cli/job.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@job_cli.command(name="plan")
@click.argument("file", type=click.Path(exists=True, dir_okay=False))
@click.option("--profile", help="Select environment overrides from the job file.")
@click.option("--output", type=click.Choice(["yaml", "json"]), default="yaml", show_default=True)
@click.option(
    "--max-cpu",
    type=click.FloatRange(min=0, min_open=True),
    help="Fail if peak CPU requests exceed this budget or are unknown.",
)
@click.option(
    "--max-memory-mib",
    type=click.FloatRange(min=0, min_open=True),
    help="Fail if peak memory requests exceed this MiB budget or are unknown.",
)
def plan(file, profile, output, max_cpu, max_memory_mib):
    """Print a redacted offline preview. Nothing is submitted or reserved."""
    job = _load(file, profile)
    try:
        result = job.plan().to_dict()
        if max_cpu is not None or max_memory_mib is not None:
            from spark_on_k8s.resource_plan import check_budget

            result["budget"] = check_budget(
                result["capacity"], max_cpu=max_cpu, max_memory_mib=max_memory_mib
            )
    except (ValueError, TypeError) as error:
        raise click.ClickException(str(error)) from error
    click.echo(
        json.dumps(result, indent=2) if output == "json" else yaml.safe_dump(result, sort_keys=False),
        nl=output == "json",
    )

    if result.get("budget", {}).get("passed") is False:
        raise click.exceptions.Exit(1)

submit(file, profile, output, timeout, receipt_file)

Submit a job using the same automation result contract as app submit.

Source code in spark_on_k8s/cli/job.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@job_cli.command(name="submit")
@click.argument("file", type=click.Path(exists=True, dir_okay=False))
@click.option("--profile", help="Select environment overrides from the job file.")
@click.option("--output", type=click.Choice(["text", "json"]), default="text", show_default=True)
@click.option(
    "--timeout",
    type=click.FloatRange(min=0, min_open=True),
    default=None,
    help="Maximum observation time in seconds; requires a wait or log waiter.",
)
@receipt_option
def submit(file, profile, output, timeout, receipt_file):
    """Submit a job using the same automation result contract as app submit."""
    from spark_on_k8s.cli.automation import execute_submission, execution

    try:
        job = JobSpec.from_file(file, profile=profile)
    except (ValueError, OSError):
        # Keep one submit document on stdout; the namespace is unknown until the job loads.
        with execution("submit", "", output) as result:
            result.update(run=None, status=None)
            raise
    execute_submission(
        job.client(), job.to_submit_kwargs(), output=output, timeout=timeout, receipt_file=receipt_file
    )

validate(file, profile, all_profiles, output)

Validate a job and its generated resources without accessing a cluster.

Source code in spark_on_k8s/cli/job.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@job_cli.command(name="validate")
@click.argument("file", type=click.Path(exists=True, dir_okay=False))
@click.option("--profile", help="Select environment overrides from the job file.")
@click.option("--all-profiles", is_flag=True, help="Check the base and every named profile offline.")
@click.option("--output", type=click.Choice(["text", "json"]), default="text")
def validate(file, profile, all_profiles, output):
    """Validate a job and its generated resources without accessing a cluster."""
    if all_profiles and profile is not None:
        raise click.UsageError("Choose --profile or --all-profiles, not both")
    try:
        if all_profiles:
            results = JobSpec.validate_profiles(file)
        else:
            JobSpec.from_file(file, profile=profile).plan()
            results = [{"profile": profile, "valid": True, "error": None}]
    except (ValueError, TypeError, OSError) as error:
        results = [{"profile": profile, "valid": False, "error": validation_error(error)}]
    valid = all(item["valid"] for item in results)
    if output == "json":
        # Like other automation results, `error` is set exactly when the exit code is nonzero.
        failure = next((item["error"] for item in results if not item["valid"]), None)
        click.echo(
            json.dumps(
                {
                    "schema_version": 1,
                    "operation": "job_validate",
                    "valid": valid,
                    "profiles": results,
                    "error": failure,
                }
            )
        )
    else:
        for item in results:
            label = "base" if item["profile"] is None else item["profile"]
            click.echo(
                f"Valid SparkJob ({label})"
                if item["valid"]
                else f"Invalid SparkJob ({label}): {item['error']['message']}",
                err=not item["valid"],
            )
    if not valid:
        raise click.exceptions.Exit(1)