SparkNotebookOperator

SparkNotebookOperator: run a git-checked-out notebook as a cluster-mode Spark job.

Example:

from airflow.providers.notebook.operators.spark_notebook import SparkNotebookOperator

run_notebook = SparkNotebookOperator(
    task_id="run_notebook",
    notebook_path="pipelines/daily_aggregate.ipynb",
    repo_url="https://dev.azure.com/dataplatformgemeenten/Dataplatform/_git/pipelines",
    repo_branch="main",
    notebook_parameters='{"run_date": "{{ ds }}"}',
    driver_memory="2g",
    executor_instances=2,
)
airflow.providers.notebook.operators.spark_notebook.DATAPROCESSING_IMAGE_ENV_VAR = 'DATAPROCESSING_IMAGE'

Set on the Airflow scheduler/dag-processor/worker pods by the platform deployment (managed-dataprocessing’s helm/airflow chart) – deliberately not a constructor parameter, so a DAG author can never point a job at an image other than the one the platform maintains.

airflow.providers.notebook.operators.spark_notebook.LAKE_GATEWAY_URL_ENV_VAR = 'DATAPLATFORM_LAKE_GATEWAY_URL'

Set on the Airflow pods by the platform deployment when the lake gateway is in use (managed-dataprocessing’s enableLakeGateway): jobs then do not mount the lake at all, and the platform’s image reaches it through the gateway as the job’s service identity. Not a constructor parameter, like the image.

class airflow.providers.notebook.operators.spark_notebook.SparkNotebookOperator(*, notebook_path: str, repo_url: str, repo_branch: str, notebook_parameters: str = '{}', driver_cores: int = 1, driver_core_limit: str = '1000m', driver_memory: str = '1g', executor_instances: int = 1, executor_cores: int = 1, executor_core_limit: str = '1000m', executor_memory: str = '1g', extra_spark_conf: dict[str, str] | None = None, identity: str | None = None, name: str | None = None, **kwargs: Any)

Bases: SparkKubernetesOperator

Run one notebook, checked out fresh from git, as a cluster-mode Spark job via the Kubeflow Spark Operator – one task, no hand-written SparkApplication manifest per pipeline.

Every field below is a distilled, load-bearing fix from getting a real notebook-as-a-scheduled-job pipeline working by hand first:

  • do_xcom_push is forced to False and not user-configurable. SparkKubernetesOperator inherits KubernetesPodOperator’s file-based xcom, which (when True) waits forever for an airflow-xcom sidecar to read /airflow/xcom/return.json from the driver pod – nothing a papermill-executed notebook writes. Live-confirmed: the task blocked 15+ minutes past the driver’s own completion before this was found and disabled.

  • The Delta session extension is repeated in every job’s own sparkConf rather than relying on the image’s baked-in spark-defaults.conf. Spark Operator generates a fresh conf ConfigMap from sparkConf and mounts it over SPARK_CONF_DIR, entirely replacing (not merging with) whatever the image ships.

  • Unity Catalog catalogs are not in sparkConf: they need a token, and the SparkApplication CR is readable. The driver gets the batch-jobs credentials from a Secret, and the image’s kernel startup (shared with JupyterHub) exchanges them and registers every catalog the batch-jobs principal can see. New MANAGED tables are UniForm (Trino-readable) by default, set by Unity Catalog itself.

  • No separate sensor task is needed alongside this operator. SparkKubernetesOperator.execute() (via KubernetesPodOperator) already blocks on the driver pod, streams its logs, and raises on a non-zero exit – the driver’s own completion IS the job’s completion. A hand-written two-task DAG that pairs this with a SparkKubernetesSensor is only polling something already waited on synchronously.

Requires the spark-driver ServiceAccount (the fixed identity every job submits under) to be able to get/list/watch/create/update on sparkapplications.sparkoperator.k8s.io and the same verbs on its /status subresource – status is a distinct RBAC resource from the parent object in Kubernetes; granting only the parent produces a 403 immediately after the SparkApplication CR is created successfully (live-confirmed: “cannot get resource sparkapplications/status”).

Almost everything about how the job runs is deliberately not a parameter here – it’s a platform-owned constant, because none of it is a decision a DAG author should be able to make differently from one notebook to the next:

  • The Spark image – read from the DATAPROCESSING_IMAGE env var, set once on the Airflow deployment (scheduler/dag-processor/worker pods) by the platform, the same image every other dataprocessing job (JupyterHub, the Azure Pipelines Spark submit step) runs.

  • The Unity Catalog identity – always the platform’s batch-jobs client, from the unity-catalog-batch-client Secret on the driver. What a job may read and write is that principal’s grants in security-center, not a per-DAG choice.

  • The lake volume (lake-shared-data at /shared) on the driver and executors.

  • The Kubernetes connection this operator submits the SparkApplication with – always kubernetes_default.

  • The namespace – always the namespace the Airflow worker itself runs in (read from the ServiceAccount projection), never a DAG-chosen one.

  • The git connection supplying checkout credentials – always git_default, the same connection the platform’s git-backed DAG bundles already use.

  • The Spark version, the ServiceAccount, imagePullSecrets, and the Secret’s name – all tied to dataprocessing_image/this platform’s own conventions, never per-notebook choices.

What is left as a parameter is genuinely per-notebook: which notebook, from where, its Papermill parameters, and how big a driver/executor it needs.

Parameters:

  • notebook_path – Path to the notebook inside the git repo, e.g. pipeline.ipynb. Templated.

  • repo_url – Git URL to clone the notebook from. Templated.

  • repo_branch – Branch/ref to check out. Templated.

  • notebook_parameters – JSON object of Papermill parameter overrides (default "{}"). Templated.

  • driver_cores – / :param driver_core_limit: / :param driver_memory: Driver sizing – defaults (1g) are sized for a light notebook, not a heavy job. Started at 512m but that OOMKilled (exit 137) on a real run the moment Delta and the storage client were loaded together for a genuine write+read cycle, not just submission – JVM + those libraries’ overhead alone exceeds a truly minimal budget.

  • executor_instances – / :param executor_cores: / :param executor_core_limit: / :param executor_memory: Executor sizing, same caveat as the driver.

  • identity – The service identity the job runs as (default: the shared batch-jobs one). With the lake gateway, a job reads and writes exactly what that identity’s team has been granted in security-center, so a job that only needs one product should run as an identity that holds only that. Its credentials are the Secret unity-catalog-batch-client-<identity>, created by the platform (one Keycloak client per identity); the name is a lowercase DNS label.

  • extra_spark_conf – Additional spark.* keys merged in after the Delta defaults above – these win on key conflicts.