Slate

Benchmarked: S3 us-east-1 → Hetzner Frankfurt · 5 ×ばつ 529 MiB model weights
 slate 984 MB/s ████████████████████████████████████
 rclone 1120 MB/s ████████████████████████████████████████ ×ばつ faster
 aws s3 cp 221 MB/s ████████ ×ばつ slower

Install

curl -fsSL https://github.com/tsushanth/slate/releases/latest/download/slate-linux-x86_64.tar.gz \
 | tar xz -C /usr/local/bin

pip install slate-sdk # Python client
pip install apache-airflow-providers-slate # Airflow operator
pip install prefect-slate # Prefect task

git clone https://github.com/tsushanth/slate
cd slate && cargo build --release

Quick start

# Start the API server
DATABASE_URL=sqlite:slate.db?mode=rwc slate-api
# Submit a transfer
curl -X POST localhost:3030/jobs \
 -H 'Content-Type: application/json' \
 -d '{"src": "s3://my-bucket/datasets/imagenet", "dst": "gs://other/datasets/imagenet"}'
# Or use the CLI (blocking, with progress bar)
slate copy s3://my-bucket/datasets/imagenet /mnt/nvme/datasets/imagenet

Features

Job queue with priority

curl -X POST localhost:3030/jobs -d '{
 "src": "s3://bucket/weights",
 "dst": "gs://other/weights",
 "priority": 10
}'

Retry with exponential backoff

curl -X POST localhost:3030/jobs -d '{
 "src": "s3://bucket/data",
 "dst": "gs://other/data",
 "max_attempts": 5
}'

Pipeline chaining

# Submit job A
JOB_A=$(curl -s -X POST localhost:3030/jobs \
 -d '{"src": "s3://raw/", "dst": "gs://stage/"}' | jq -r .id)
# Job B won't start until job A completes
curl -X POST localhost:3030/jobs -d "{
 \"src\": \"gs://stage/\",
 \"dst\": \"gs://prod/\",
 \"depends_on\": \"$JOB_A\"
}"

Cron scheduling

# Daily at 2am UTC
curl -X POST localhost:3030/crons -d '{
 "src": "s3://data-lake/raw/",
 "dst": "gs://ml-staging/raw/",
 "cron": "0 2 * * *"
}'
# Every 6 hours
curl -X POST localhost:3030/crons -d '{
 "src": "s3://checkpoints/latest/",
 "dst": "/mnt/nvme/checkpoints/",
 "cron": "0 */6 * * *"
}'
# List / delete schedules
curl localhost:3030/crons
curl -X DELETE localhost:3030/crons/<id>

Webhook callbacks

curl -X POST localhost:3030/jobs -d '{
 "src": "s3://bucket/data",
 "dst": "gs://other/data",
 "callback_url": "https://your-service/hooks/slate"
}'
# POST fires on completion or terminal failure with job metadata

Cost tracking

curl localhost:3030/cost # aggregate across all completed jobs
curl localhost:3030/jobs/<id>/cost # per-job egress cost estimate

Cancel queued jobs

curl -X POST localhost:3030/jobs/<id>/cancel

Real-time progress (SSE)

curl localhost:3030/jobs/<id>/events
# data: {"job_id":"...","bytes_transferred":1073741824,"bytes_total":2684354560,"throughput_mbps":512.3}

Airflow integration

pip install apache-airflow-providers-slate

from apache_airflow_providers_slate.operators.slate import SlateTransferOperator
with DAG("ml_pipeline", schedule="@daily") as dag:
 ingest = SlateTransferOperator(
 task_id="ingest_dataset",
 src="s3://raw-data/datasets/imagenet/",
 dst="gs://ml-staging/datasets/imagenet/",
 )
 train = SlateTransferOperator(
 task_id="copy_weights",
 src="gs://ml-staging/weights/llama-3/",
 dst="/mnt/nvme/weights/",
 priority=10,
 max_attempts=5,
 )
 ingest >> train

Prefect integration

pip install prefect-slate

from prefect import flow
from prefect_slate import slate_transfer, SlateCredentials
creds = SlateCredentials(base_url="http://slate-api:3030")
@flow
def ml_pipeline():
 result = slate_transfer(
 src="s3://raw/datasets/imagenet/",
 dst="gs://staging/datasets/imagenet/",
 slate_credentials=creds,
 )
 print(f"Transferred {result['bytes_transferred']} bytes at {result['peak_throughput_mbps']:.0f} MB/s")

API reference

Supported stores

Configuration

Architecture

What's next