Skip to main content

Multi-Step Pipelines with Luigi Dockerization

While Solti natively schedules and executes single-container workloads (jobs) and periodic triggers (cronjobs), multi-step Directed Acyclic Graph (DAG) pipelines — complete with step dependencies, artifact checking, and checkpointing — can be implemented using Spotify's Luigi.

By containerizing a Luigi pipeline, the entire multi-task workflow runs as a self-contained, cost-optimized Solti workload without the operational burden of hosting external orchestrators like Apache Airflow or Argo Workflows.

Step 1: Define Tasks in Python (workflow.py)​

Define tasks, dependencies, and outputs using standard Luigi classes:

import luigi
import os

class StepOne(luigi.Task):
    output_path = luigi.Parameter(default="/mnt/data/step_one_output.txt")

    def output(self):
        return luigi.LocalTarget(self.output_path)

    def run(self):
        os.makedirs(os.path.dirname(self.output_path), exist_ok=True)
        with self.output().open('w') as f:
            f.write('Step One Completed\n')


class StepTwo(luigi.Task):
    output_path = luigi.Parameter(default="/mnt/data/step_two_output.txt")

    def requires(self):
        return StepOne()

    def output(self):
        return luigi.LocalTarget(self.output_path)

    def run(self):
        with self.output().open('w') as f:
            f.write('Step Two Completed\n')


if __name__ == '__main__':
    luigi.build([StepTwo()], local_scheduler=True)

Step 2: Package the Pipeline into Docker (Dockerfile)​

Package your workflow script with its runtime dependencies:

FROM python:3.13-slim

WORKDIR /app
RUN pip install --no-cache-dir luigi

COPY workflow.py /app/workflow.py

ENTRYPOINT ["python", "workflow.py"]

Build and push the image to your container registry (e.g. AWS ECR, GCP Artifact Registry, Azure CR, GHCR, or Docker Hub).

Step 3: Submit Workload with Persistent Cloud Storage​

Mount your cloud storage bucket via Solti FUSE volumes so intermediate task outputs and final deliverables persist across runs.

Option A: Submit via CLI​

solti job run \
  --image your-registry.example.com/luigi-workflow:latest \
  --volume my-cloud-storage:/mnt/data \
  --resources cpu=4,memory=8Gi \
  --tag etl-pipeline

Option B: Submit via Declarative YAML (workflow-job.yaml)​

name: luigi-data-pipeline
image: your-registry.example.com/luigi-workflow:latest
volumes:
  data:                    # Volume name (the map key)
    bucket: my-cloud-storage
    mountPath: /mnt/data
resources:
  cpu: 4
  memory: 8Gi
tags:
  - etl-pipeline

Submit via:

solti job run -f workflow-job.yaml

Step 4: Automate Execution on Schedule (CronJob)​

If the multi-step pipeline should run on a recurring schedule (e.g., daily at 2:00 AM UTC), register it as a Solti CronJob:

solti cronjob create \
  --name daily-luigi-pipeline \
  --schedule "0 2 * * *" \
  --file workflow-job.yaml

Solti will automatically trigger and run the containerized pipeline on the defined schedule. See the Cronjobs API for the REST equivalent.