Metadata-Version: 2.4
Name: vigie-spark-operator
Version: 0.1.2
Summary: Airflow operator that submits Kubeflow SparkApplications on Kubernetes (AF2 + AF3)
Author: HEMA Tiemoko Thierry
License: MIT
Project-URL: Homepage, https://github.com/Jaali-co/vigie-spark-operator
Project-URL: Changelog, https://github.com/Jaali-co/vigie-spark-operator/blob/main/CHANGELOG.md
Project-URL: Issues, https://github.com/Jaali-co/vigie-spark-operator/issues
Keywords: airflow,spark,kubernetes,spark-operator
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: Apache Airflow
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: System :: Distributed Computing
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: kubernetes>=28.1.0
Provides-Extra: airflow
Requires-Dist: apache-airflow<4,>=2.7; extra == "airflow"
Provides-Extra: dev
Requires-Dist: pytest>=7.4; extra == "dev"
Requires-Dist: pytest-cov>=4.1; extra == "dev"
Requires-Dist: build>=1.2; extra == "dev"
Requires-Dist: twine>=5.0; extra == "dev"
Dynamic: license-file

# vigie-spark-operator

Open-source Airflow operator that submits a Kubeflow **SparkApplication**
(Spark Operator on Kubernetes). Works with **Airflow 2.x and 3.x**.

Maintainer: **HEMA Tiemoko Thierry** · License: **MIT**

## Requirements

| | |
|---|---|
| Python | `>=3.9` |
| Runtime dep | `kubernetes>=28.1.0` |
| Airflow (runtime) | `>=2.7,<4` — provided by the worker image, or install extra `[airflow]` |
| Cluster | [Kubeflow Spark Operator](https://github.com/kubeflow/spark-operator) CRD `SparkApplication` installed; RBAC for the Airflow worker SA |

Declared in `pyproject.toml`. Dev tools: see `requirements-dev.txt`.

```bash
pip install vigie-spark-operator
# or with Airflow pinned as dependency (usually not needed inside an AF image)
pip install "vigie-spark-operator[airflow]"
```

## Quick start

```python
from vigie_spark_operator import SparkK8sOperator

SparkK8sOperator(
    task_id="spark_pi",
    name="pi",
    namespace="spark-jobs",
    image="spark:3.5.1",
    application_type="Scala",
    main_class="org.apache.spark.examples.SparkPi",
    main_application_file="local:///opt/spark/examples/jars/spark-examples_2.12-3.5.1.jar",
    arguments=["10"],
    spark_version="3.5.1",
    service_account="spark",
    dry_run=False,
)
```

### Optional Vigie capacity labels

Generic by default (no `managed-by`). For Vigie Supervision, pass
`managed_by="vigie"` so driver/executor get `managed-by=vigie` plus
`dag_id` / `task_id` / `run_id`:

```python
SparkK8sOperator(..., managed_by="vigie")
```

Disable context labels entirely: `inject_airflow_labels=False`.

## Install on Airflow workers

Preferred: bake the wheel into the worker image.

Alternative: `_PIP_ADDITIONAL_REQUIREMENTS=vigie-spark-operator==0.1.2` (slower, less reproducible).

AF2 / AF3: the package selects `airflow.models` (AF2) or `airflow.sdk` (AF3) via `compat.py`.

## Main parameters

| Parameter | Default | Notes |
|-----------|---------|--------|
| `name` | *(required)* | Base name for the SparkApplication |
| `namespace` | *(required)* | Target K8s namespace |
| `image` | *(required)* | Spark runtime image |
| `main_application_file` | *(required)* | App entry (`local://…`, `s3a://…`, …) |
| `application_type` | `"Python"` | `Python` / `Scala` / `Java` / `R` |
| `main_class` | `None` | Required for Scala/Java |
| `arguments` | `[]` | App args |
| `spark_version` | `"3.1.1"` | Reported to the Spark Operator |
| `driver_cores` / `driver_memory` | `1` / `"512m"` | Driver resources |
| `driver_cores_request` | `"20m"` | K8s CPU request |
| `executor_cores` / `executor_memory` | `1` / `"512m"` | Executor resources |
| `executor_cores_request` | `"400m"` | K8s CPU request |
| `executor_instances` | `2` | Number of executors |
| `service_account` | `"spark"` | Pod SA in the job namespace |
| `spark_conf` | `{}` | Extra Spark conf |
| `labels` / `env` / `jars` | `{}` / `{}` / `[]` | Merged labels, env vars, extra jars |
| `pvc_configs` / `emptydir_configs` | `[]` | Volume mounts (`PvcConfig` / `EmptyDirConfig` or dicts) |
| `run_as_user` / `fs_group` | `0` / `0` | Shared security context |
| `allow_privilege_escalation` | `False` | Shared; optional `driver_*` / `executor_*` overrides |
| `timeout_job` | `3600` | Job timeout (s); graceful delete then force-kill |
| `timeout` | `None` | Legacy alias → sets `timeout_job` |
| `poll_interval` | `10` | Status poll (s) |
| `api_error_retries` | `5` | Retries on K8s `429` / `503` |
| `in_cluster` | `True` | Use in-cluster kubeconfig |
| `fail_on_unschedulable` | `True` | Fail if pods stay Pending |
| `dry_run` | `False` | Build + log manifest, no submit |
| `inject_airflow_labels` | `True` | Inject `dag_id` / `task_id` / `run_id` |
| `managed_by` | `None` | e.g. `"vigie"` for capacity collector |

Also accepts standard `BaseOperator` kwargs (`task_id`, `queue`, `executor`, …).

## Behaviour highlights

- Labels on **metadata, driver and executor**
- `dry_run=True`: validates resources summary, exits 0 without creating the CR
- Exponential backoff on transient K8s API errors (`429`, `503`)
- Near `timeout_job`: delete SparkApplication, then force-kill leftover pods

## Development

```bash
python3 -m venv .venv && source .venv/bin/activate
pip install -r requirements-dev.txt
pip install -e .
pytest -q
# packaging smoke (build → twine check → clean install)
../scripts/test_pypi_packaging.sh .
```

## Links

- Changelog: [CHANGELOG.md](CHANGELOG.md)
- Issues: https://github.com/Jaali-co/vigie-spark-operator/issues
