Run Spark Jobs with Airflow

The IOMETE Airflow Plugin is an extension that lets you trigger and manage Spark jobs from your Airflow workflows. It submits job runs, monitors their status, and supports cancellation on task kill.

Installation

Prerequisites

• Python 3.10 – 3.13
• Apache Airflow >=2.10.5, <4.0.0 (both Airflow 2.x and 3.x are supported)

Setup Airflow (Optional)

note

Skip this step if you already have Airflow installed and configured.

Two ways to install Airflow with the IOMETE plugin locally:

1. Docker Setup Clone the Airflow plugin repository and run docker-compose. It builds the latest version from source and starts Airflow webserver on http://localhost:8080. May take a couple of minutes to start.

docker-compose up --build

2. Helm Install Install Airflow on Kubernetes using the official Helm chart. Update the docker image in values.yaml to use the IOMETE Airflow image matching your Airflow version. This runs Airflow with the IOMETE plugin pre-installed.

Manually Install IOMETE Airflow Plugin

note

Skip this step if you installed Airflow with one of the methods above.

If you already have Airflow installed, install the plugin manually. In your Airflow home directory, run:

pip install iomete-airflow-plugin

Restart the Airflow webserver after installation.

Configuration

The operator needs API credentials and endpoint details to communicate with IOMETE. You pass these directly to IometeOperator (or share them across tasks via the DAG's default_args):

• host — IOMETE platform host URL. Example: https://sandbox.iomete.cloud
• domain — IOMETE domain identifier.
• A personal access token, supplied in one of two ways (see below).

Because credentials are per-task, a single Airflow instance can trigger jobs against multiple IOMETE clusters by passing different host/domain/token values per task.

Providing the access token

Supply exactly one of these two parameters:

• access_token — the raw token as a string. This field is not templatable: Airflow persists rendered template fields to its metadata database and shows them in the UI's "Rendered Template" tab, so templating a raw token would leak it.
• access_token_variable — the name of an Airflow Variable holding the token. The plugin calls Variable.get at task-execute time, so the raw token never appears in the DAG code or rendered fields. This is the recommended approach.

To use access_token_variable, first store your token as an Airflow Variable. From the Airflow UI, navigate to Admin > Variables and add a Variable (for example, named iomete_access_token). We recommend prefixing Variable names with iomete_ to keep them grouped.

If the resolved Variable is unset or empty at execute time, the task fails with a clear error.

Multiple IOMETE clusters

Store one Variable per target cluster (e.g. iomete_prod_token, iomete_dev_token) and reference each by name from the relevant task's access_token_variable.

Usage

Once the plugin is installed and configured, you can use the IometeOperator in your DAGs to trigger Spark jobs.

All DAGs should be placed in the dags/ folder in your Airflow home directory (or the directory where you ran docker-compose).

Operator Parameters

Parameter	Type	Required	Default	Description
task_id	str	Yes	—	Unique identifier for the task in the DAG.
job_id	str	Yes	—	Spark Job ID (UUID) or name from IOMETE platform.
host	str	Yes	—	IOMETE platform host URL.
domain	str	Yes	—	IOMETE domain identifier.
access_token	str	One of these two	—	Personal access token, passed as a raw string. Not templatable.
access_token_variable	str	One of these two	—	Name of an Airflow Variable holding the token. Resolved via Variable.get at execute time.
host_verify	bool	No	True	Verify the TLS certificate of the IOMETE host. Set to False to disable.
config_override	dict or str	No	{}	Configuration overrides for the Spark Job (see below).
polling_period_seconds	int	No	10	Interval in seconds between status checks while the job is running.
do_xcom_push	bool	No	False	Push job_id and job_run_id to XCom for downstream tasks.

Specify either access_token or access_token_variable, never both. The job_id, config_override, host, domain, and access_token_variable fields are Jinja-templatable; access_token is not.

Config Override

The config_override parameter accepts a dict (or JSON string) with these optional fields:

{
  "arguments": ["arg1", "arg2"],
  "envVars": {
    "key": "value"
  },
  "sparkConf": {
    "spark.example.variable": "sample_value"
  }
}

Job Run States

The operator monitors the job run and reports these states:

State	Description	Final?
ENQUEUED	Job is queued	No
SUBMITTED	Job is being deployed (~1 min)	No
RUNNING	Job is executing	No
COMPLETED	Job finished successfully	Yes
FAILED	Job failed	Yes
ABORTED	Job was cancelled	Yes
ABORTING	Job cancellation in progress	No

The operator raises an AirflowException if the job ends in FAILED or ABORTED state. If the Airflow task is killed, the operator automatically cancels the running job in IOMETE.

Example DAGs

Link to Source

Source of examples below can be found in the GitHub repository.

Single Task

import pendulum
from airflow import DAG
from iomete_airflow_plugin.iomete_operator import IometeOperator

args = {
    "owner": "airflow",
    "email": ["airflow@example.com"],
    "depends_on_past": False,
    "start_date": pendulum.today("UTC"),
}

dag = DAG(dag_id="iomete-task", default_args=args, schedule=None)

# Resolves the token from the Airflow Variable "iomete_access_token" at execute time.
task = IometeOperator(
    task_id="iomete-catalog-sync-task",
    job_id="0761a510-3a66-4c72-b06e-9d071f30d85d",
    host="https://YOUR.iomete.host",
    domain="YOUR_DOMAIN",
    access_token_variable="iomete_access_token",
    dag=dag,
)

Task with Config Overrides

Overriding Config

You can dynamically change config_override params with each run by choosing "Run w/ config" from Airflow's UI.

import pendulum
from airflow import DAG
from iomete_airflow_plugin.iomete_operator import IometeOperator

args = {
    "owner": "airflow",
    "email": ["airflow@example.com"],
    "depends_on_past": False,
    "start_date": pendulum.today("UTC"),
}

dag = DAG(
    dag_id="iomete-task-with-args",
    default_args=args,
    schedule=None,
    params={
        "job_id": "11ef35a6-ff9d-4996-bf16-7a7ef0baf4fc",
        "config_override": {
            "envVars": {"env1": "value1"},
            "arguments": ["arg1"],
            "sparkConf": {"spark.example.variable": "sample_value"},
        },
    },
)

task = IometeOperator(
    task_id="iomete-catalog-sync-task-with-config",
    job_id="{{ params.job_id }}",
    config_override="{{ params.config_override }}",
    host="https://YOUR.iomete.host",
    domain="YOUR_DOMAIN",
    access_token_variable="iomete_access_token",
    dag=dag,
)

Sequential Execution

import pendulum
from airflow import DAG
from iomete_airflow_plugin.iomete_operator import IometeOperator

# Shared IOMETE connection details on the DAG so each task does not repeat them.
args = {
    "owner": "airflow",
    "email": ["airflow@example.com"],
    "depends_on_past": False,
    "start_date": pendulum.today("UTC"),
    "host": "https://YOUR.iomete.host",
    "domain": "YOUR_DOMAIN",
    "access_token_variable": "iomete_access_token",
}

dag = DAG(dag_id="iomete-demo", default_args=args, schedule=None)

catalog_task = IometeOperator(
    task_id="task-01-sync-catalog",
    job_id="iomete-catalog-sync",  # using job name
    dag=dag,
)

sql_task = IometeOperator(
    task_id="task-02-run-sql",
    job_id="sql-runner",  # using job name
    dag=dag,
)

# sql_task runs first, then catalog_task
sql_task >> catalog_task

Using XCom to Pass Data Between Tasks

When do_xcom_push=True, the operator pushes job_id and job_run_id to XCom, allowing downstream tasks to reference the job run.

task = IometeOperator(
    task_id="my-task",
    job_id="my-spark-job",
    host="https://YOUR.iomete.host",
    domain="YOUR_DOMAIN",
    access_token_variable="iomete_access_token",
    do_xcom_push=True,
    dag=dag,
)

Downstream tasks can then access the values:

job_run_id = "{{ ti.xcom_pull(task_ids='my-task', key='job_run_id') }}"
job_id = "{{ ti.xcom_pull(task_ids='my-task', key='job_id') }}"

Migrating from 2.x

The 2.x plugin read connection details from four Airflow Variables (iomete_host, iomete_access_token, iomete_domain, iomete_host_verify) and exposed a variable_prefix parameter to namespace them. Both were removed in 3.0.0. Pass host, domain, and either access_token or access_token_variable (and optionally host_verify) directly to IometeOperator. If you previously used variable_prefix, set access_token_variable to the full Airflow Variable name instead (for example, access_token_variable="iomete_prod_token").

Support

For support and assistance, use the IOMETE platform's support section or contact the IOMETE support team at support@iomete.com.

Resources

API Token

Read our guide on how to create a personal access token for API

Github Repo

Check out Github repo for more information & examples