apache-airflow-providers-slate 0.2.0

Apache Airflow provider for Slate — high-throughput multi-cloud data transfers

apache-airflow-providers-slate

Apache Airflow provider for Slate — move datasets and model weights between object stores from your Airflow DAGs at 984 MB/s, 4.4× faster than aws s3 cp.

Install

pip install apache-airflow-providers-slate

Setup

Add a connection in the Airflow UI (or via env var):

Connection ID:   slate_default
Connection Type: HTTP
Host:            http://your-slate-api-host
Port:            3030

Or via environment variable:

AIRFLOW_CONN_SLATE_DEFAULT='{"conn_type": "http", "host": "http://localhost", "port": 3030}'

Make sure slate-api is running:

DATABASE_URL=sqlite:slate.db?mode=rwc slate-api

Usage

from airflow import DAG
from airflow.utils.dates import days_ago
from apache_airflow_providers_slate.operators.slate import SlateTransferOperator

with DAG(
    dag_id="ml_data_pipeline",
    schedule="@daily",
    start_date=days_ago(1),
    catchup=False,
) as dag:

    # Ingest raw dataset from S3 to GCS staging
    ingest = SlateTransferOperator(
        task_id="ingest_dataset",
        src="s3://raw-data/datasets/imagenet/",
        dst="gs://ml-staging/datasets/imagenet/",
    )

    # Copy model weights to GPU node — runs after ingest
    copy_weights = SlateTransferOperator(
        task_id="copy_weights_to_gpu",
        src="gs://ml-staging/weights/llama-3-70b/",
        dst="/mnt/nvme/weights/llama-3-70b/",
        priority=10,          # pick up before lower-priority jobs
        max_attempts=5,       # retry up to 5 times
        poll_interval=10.0,
    )

    ingest >> copy_weights

The operator returns job metadata via XCom so downstream tasks can reference it:

def use_result(**context):
    result = context["ti"].xcom_pull(task_ids="ingest_dataset")
    print(f"Transferred {result['bytes_transferred']} bytes")
    print(f"Peak speed: {result['peak_throughput_mbps']:.1f} MB/s")

Supported stores

URL scheme	Provider
s3://bucket/prefix	AWS S3, MinIO, Cloudflare R2
gs://bucket/prefix	Google Cloud Storage
az://container/prefix	Azure Blob Storage
/path or file:///path	Local filesystem

Any combination of source and destination works.

Operator reference

SlateTransferOperator(
    task_id="...",
    src="s3://...",             # required
    dst="gs://...",             # required
    slate_conn_id="slate_default",  # Airflow connection ID
    priority=0,                 # higher = picked up sooner
    max_attempts=3,             # retries with exponential backoff
    poll_interval=5.0,          # seconds between status polls
    transfer_timeout=None,      # raise after N seconds (None = no limit)
)